From 0b391f49a25f9923b348bb26157e99d78d2eed4a Mon Sep 17 00:00:00 2001 From: jochen Date: Sun, 18 Oct 2026 19:10:05 +0200 Subject: [PATCH 1/3] Write each module's have and the node's own facts beside the report, grouped per module and keyed by the module the controller names (issue 1132) The node-engine reads a resource's module and a declaration's builds, says it does in every report (reads_modules), groups the apply's outcome per module with every failing resource named, and writes have.. and have..node to the bucket only when a value differs, never ahead of or in place of the report (novox/hq ADR 0429 section 4). --- cmd/mesh-host/main.go | 31 ++- cmd/mesh-host/module_facts_test.go | 38 +++ go.mod | 2 +- go.sum | 4 +- internal/apply/apply.go | 40 ++- internal/apply/module_results.go | 99 +++++++ internal/apply/module_results_test.go | 76 +++++ internal/apply/move_out.go | 2 +- internal/declaration/declaration.go | 142 ++++++++-- internal/declaration/module_test.go | 60 ++++ internal/link/bus.go | 45 +++ internal/link/have.go | 219 +++++++++++++++ internal/link/have_nats_integration_test.go | 104 +++++++ internal/link/have_test.go | 291 ++++++++++++++++++++ internal/link/hearing_nats.go | 9 + internal/link/messages.go | 19 ++ internal/link/queue.go | 21 ++ internal/link/queue_test.go | 23 ++ internal/liveness/liveness.go | 13 +- 19 files changed, 1193 insertions(+), 45 deletions(-) create mode 100644 cmd/mesh-host/module_facts_test.go create mode 100644 internal/apply/module_results.go create mode 100644 internal/apply/module_results_test.go create mode 100644 internal/declaration/module_test.go create mode 100644 internal/link/have.go create mode 100644 internal/link/have_nats_integration_test.go create mode 100644 internal/link/have_test.go diff --git a/cmd/mesh-host/main.go b/cmd/mesh-host/main.go index 06e1a8b..7a49a5d 100644 --- a/cmd/mesh-host/main.go +++ b/cmd/mesh-host/main.go @@ -1663,7 +1663,7 @@ func declaredResolvConf(d *declaration.Declaration) (string, string, bool) { if !ok || f.Path != network.ResolvConf || f.Seeded() || f.Into != "" { continue } - module, ok := liveness.ModuleOf(f.ID) + module, ok := d.ModuleOfID(f.ID) if !ok { return "", "", false } @@ -2323,6 +2323,9 @@ func applyAndKeepHeld(ctx context.Context, opts options, raw []byte, signed *sto report.Failed = map[string]string{"apply": applyErr.Error()} } report.Failed = withUnanswered(report.Failed, probe) + // And how each module of the declaration came out, for the have keys written beside this report (novox/hq ADR 0429 + // §4, issue 1132): not on the wire. + report.Modules = moduleFacts(declared, outcome, applyErr) // Which resources did not apply, for whether a reconcile's account is news (issue 233): not said on the wire. report.Unapplied = unapplied(declared, outcome, applyErr) return report @@ -2575,3 +2578,29 @@ func unhealthyWords(r liveness.State) string { return fmt.Sprintf("%s (%s %s) is unhealthy: %s, %d automatic restart(s) since this engine began watching it", r.ID, r.Kind, r.Target, r.Reason, r.Restarts) } + +// moduleFacts is how each module of a declaration came out of an apply, with the build the controller sent for it +// (novox/hq ADR 0429 §4, issue 1132): sorted by module so the facts do not move between applies. A module an adopted +// node has not taken is held as found and not applied, so it is no fact. +func moduleFacts(d *declaration.Declaration, outcome apply.Report, applyErr error) []link.ModuleFact { + if d == nil { + return nil + } + results := apply.ModuleResults(d, outcome, applyErr) + facts := make([]link.ModuleFact, 0, len(results)) + for module, result := range results { + if _, untaken := untakenModules(d)[module]; untaken { + continue + } + facts = append(facts, link.ModuleFact{Module: module, Build: d.Builds[module], Failed: result.Failed, Detail: result.Detail}) + } + sort.Slice(facts, func(i, j int) bool { return facts[i].Module < facts[j].Module }) + return facts +} + +func untakenModules(d *declaration.Declaration) map[string][]string { + if d.Adoption == nil { + return nil + } + return d.Adoption.Untaken +} diff --git a/cmd/mesh-host/module_facts_test.go b/cmd/mesh-host/module_facts_test.go new file mode 100644 index 0000000..79b485a --- /dev/null +++ b/cmd/mesh-host/module_facts_test.go @@ -0,0 +1,38 @@ +package main + +import ( + "errors" + "testing" + + "github.com/novox/mesh-host/internal/apply" + "github.com/novox/mesh-host/internal/declaration" +) + +// 1132/AC1, AC3: the report carries one fact per module the declaration names, with the build the controller sent for +// it; a module not yet taken on an adopted node is not one. +func TestTheReportCarriesOneFactPerModuleWithItsBuild(t *testing.T) { + d, err := declaration.Parse([]byte(`{"declaration":1,"builds":{"a":"b-1","b":"b-2"},"resources":[ + {"id":"a.dir","type":"directory","path":"/srv/a","module":"a"}, + {"id":"b.dir","type":"directory","path":"/srv/b","module":"b"}]}`)) + if err != nil { + t.Fatal(err) + } + failure := &apply.Error{Resource: "b.dir", Err: errors.New("no space"), Failures: []apply.Failure{{Resource: "b.dir", Err: errors.New("no space")}}} + facts := moduleFacts(d, apply.Report{}, failure) + if len(facts) != 2 || facts[0].Module != "a" || facts[0].Build != "b-1" || facts[0].Failed || + facts[1].Module != "b" || facts[1].Build != "b-2" || !facts[1].Failed || facts[1].Detail == "" { + t.Fatalf("facts %+v", facts) + } + adopted, err := declaration.Parse([]byte(`{"declaration":1,"adoption":{"taken":["a"],"untaken":{"b":["b.dir"]}},"resources":[ + {"id":"a.dir","type":"directory","path":"/srv/a","module":"a"}, + {"id":"b.dir","type":"directory","path":"/srv/b","module":"b"}]}`)) + if err != nil { + t.Fatal(err) + } + if facts := moduleFacts(adopted, apply.Report{}, nil); len(facts) != 1 || facts[0].Module != "a" { + t.Fatalf("an untaken module is a fact: %+v", facts) + } + if moduleFacts(nil, apply.Report{}, nil) != nil { + t.Fatal("facts without a declaration") + } +} diff --git a/go.mod b/go.mod index 406e5ba..26d0523 100644 --- a/go.mod +++ b/go.mod @@ -3,7 +3,7 @@ module github.com/novox/mesh-host go 1.26.0 require ( - git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261015210729-d3e182503c19 + git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261018164924-2b420603fe83 github.com/nats-io/nats.go v1.54.0 golang.org/x/crypto v0.57.0 golang.org/x/sys v0.48.0 diff --git a/go.sum b/go.sum index 007020d..e14ad38 100644 --- a/go.sum +++ b/go.sum @@ -1,5 +1,5 @@ -git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261015210729-d3e182503c19 h1:mAMt3KFmrHzDiBv4MLL94opWcG9y73ISMrBUyisapQE= -git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261015210729-d3e182503c19/go.mod h1:GFuZUElBZ9A++mxgIKo97aXXo+kV0uJ/UkbhQPPIbrY= +git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261018164924-2b420603fe83 h1:BrHyE3ifFMUQLIza6I/iNlEgdcysB9/Y+1FO6Q9xws8= +git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261018164924-2b420603fe83/go.mod h1:GFuZUElBZ9A++mxgIKo97aXXo+kV0uJ/UkbhQPPIbrY= github.com/klauspost/compress v1.20.0 h1:a3C1ke2ohxFymNlb2HWAHjDeKCI90scRskErZkR0ezA= github.com/klauspost/compress v1.20.0/go.mod h1:LUdAzn7YLVvxLpc7y3V1m40wESHTgc1422pwwBSKYuI= github.com/nats-io/nats.go v1.54.0 h1:vsXoOxjHp/GmPUN+EcI7uOf/uB+iAP+kEsAFNQN0yzA= diff --git a/internal/apply/apply.go b/internal/apply/apply.go index bf24228..5eaa28b 100644 --- a/internal/apply/apply.go +++ b/internal/apply/apply.go @@ -179,6 +179,23 @@ type Error struct { // (novox/hq ADR 0136) or a directory the engine refused (issue 495). Named, because "everything // was attempted" is then untrue, and a report that says so after skipping a module lies. Modules []string + + // Failures is every resource that failed, the first among them, each with its error (issue 1132, novox/hq ADR + // 0429 §4): Others counts the rest and names none, and a module's state on the bus names what failed in it. + Failures []Failure +} + +// Failure is one resource that failed in an apply, and why. +type Failure struct { + Resource string + Err error +} + +// ModuleResult is how one module's part of an apply ended (issue 1132). +type ModuleResult struct { + Failed bool + // Detail names each failing resource with its error, or why the module was not attempted. + Detail string } func (e *Error) Error() string { @@ -747,6 +764,7 @@ func ApplyMindingWindows( first := failures[0] first.Done = report first.Others = len(failures) - 1 + first.Failures = failuresOf(failures) first.Modules = gatedModules() log(" kept " + guardPrefix + "*: the guard is not up, so what it replaces was left in force") return report, known, first @@ -758,7 +776,7 @@ func ApplyMindingWindows( // **A module whose step did not complete is skipped from there on** (novox/hq ADR 0136). // Reported rather than passed over in silence: "not attempted" and "nothing to do" are // different answers, and only one of them is somebody's to fix. - if module, ours := moduleOf(resource.Identity()); ours && gated[module] { + if module, ours := d.ModuleOfID(resource.Identity()); ours && gated[module] { report.Outcomes = append(report.Outcomes, Outcome{ ID: resource.Identity(), Type: string(resource.Kind()), Target: resource.Target(), Action: "skipped", Detail: gatedWhy[module], @@ -920,7 +938,7 @@ func ApplyMindingWindows( in.refused[filepath.Clean(dir.Path)] = err } if dir, ok := resource.(*declaration.Directory); ok && in.refused[filepath.Clean(dir.Path)] != nil { - if module, ours := moduleOf(resource.Identity()); ours { + if module, ours := d.ModuleOfID(resource.Identity()); ours { gated[module] = true gatedWhy[module] = fmt.Sprintf("its directory %s was refused (novox/hq issue 495), so "+ "nothing more of this module was attempted", filepath.Clean(dir.Path)) @@ -969,7 +987,7 @@ func ApplyMindingWindows( // // An action still gates the machine: the bootstrap is a row of them, each making the // next possible, and they belong to no module. - if module, ours := moduleOf(resource.Identity()); ours { + if module, ours := d.ModuleOfID(resource.Identity()); ours { gated[module] = true gatedWhy[module] = "a step this module declares did not complete" log(fmt.Sprintf(" gated %s.*: a step it declares did not complete, so the rest "+ @@ -979,6 +997,7 @@ func ApplyMindingWindows( failed.Stopped = true failed.Done = report failed.Others = len(failures) - 1 + failed.Failures = failuresOf(failures) failed.Modules = gatedModules() return report, known, failed } @@ -1162,6 +1181,7 @@ func ApplyMindingWindows( first := failures[0] first.Done = report first.Others = len(failures) - 1 + first.Failures = failuresOf(failures) first.Modules = gatedModules() return report, known, first } @@ -3696,16 +3716,10 @@ func unitKey(scope, user, unit string) string { // False for what the mesh declares in its own right: the foundation's resources carry no dot at all, // and the adoption's are named for the mesh rather than for a module. Both belong to no module, and // their gate is therefore the machine's. -func moduleOf(identity string) (string, bool) { - if strings.HasPrefix(identity, declaration.AdoptionPrefix) { - return "", false - } - at := strings.LastIndex(identity, ".") - if at <= 0 { - return "", false - } - return identity[:at], true -} +// +// This is the guess from an id alone, for a caller with no declaration in hand; a caller with one asks +// Declaration.ModuleOfID, which reads the module the controller said (issue 1132). +func moduleOf(identity string) (string, bool) { return declaration.ModuleGuess(identity) } // handOver removes what a process replaces, once the process is running and still is a moment later // (novox/hq issue 213, ADR 0184). A replacement that is not up keeps what it replaces in place diff --git a/internal/apply/module_results.go b/internal/apply/module_results.go new file mode 100644 index 0000000..372ac51 --- /dev/null +++ b/internal/apply/module_results.go @@ -0,0 +1,99 @@ +package apply + +import ( + "errors" + "fmt" + "sort" + "strings" + + "github.com/novox/mesh-host/internal/declaration" +) + +// failuresOf is every failure of an apply as the error carries it: the first among them, then the rest, so a module +// whose resource failed second is still named (issue 1132, and issue 500, which counted the rest and named none). +func failuresOf(failures []*Error) []Failure { + out := make([]Failure, 0, len(failures)) + for _, f := range failures { + out = append(out, Failure{Resource: f.Resource, Err: f.Err}) + } + return out +} + +// ModuleResults groups an apply's outcome per module the declaration names (novox/hq ADR 0429 §4, issue 1132): the +// module's resources all applied, or it failed and Detail says which resource failed with its error, which resources +// were skipped behind a step that did not complete, or that its resources were not attempted because a step of no +// module stopped the apply. A module is named by what the declaration says of each resource (ModuleOfID), so one +// whose id would guess another is still its own. +// +// A failure that is no resource's (an error that is not an Error, or a resource of no module) belongs to no module and +// changes none: the node says it in its report. The report is the outcome of an apply that did not fail; an apply that +// failed carries its own account in the Error, and that is read in its place. +func ModuleResults(d *declaration.Declaration, report Report, err error) map[string]ModuleResult { + if d == nil { + return nil + } + out := map[string]ModuleResult{} + attempted := map[string]bool{} + var failed *Error + errors.As(err, &failed) + outcomes := report.Outcomes + if failed != nil { + outcomes = append(append([]Outcome(nil), outcomes...), failed.Done.Outcomes...) + } + skipped := map[string]string{} + for _, o := range outcomes { + module, ok := d.ModuleOfID(o.ID) + if !ok { + continue + } + if o.Action == "skipped" { + skipped[module] = o.Detail + continue + } + attempted[o.ID] = true + } + var reasons = map[string][]string{} + for _, r := range d.Resources { + if module, ok := d.ModuleOfID(r.Identity()); ok { + out[module] = ModuleResult{} + if failed != nil && failed.Stopped && !attempted[r.Identity()] { + reasons[module] = append(reasons[module], fmt.Sprintf("%s was not attempted: %s stopped the apply", r.Identity(), failed.Resource)) + } + } + } + if failed != nil { + for _, f := range failed.Failures { + if module, ok := d.ModuleOfID(f.Resource); ok { + if _, named := out[module]; named { + reasons[module] = append(reasons[module], fmt.Sprintf("%s failed: %v", f.Resource, f.Err)) + } + } + } + if len(failed.Failures) == 0 { + if module, ok := d.ModuleOfID(failed.Resource); ok { + if _, named := out[module]; named { + reasons[module] = append(reasons[module], fmt.Sprintf("%s failed: %v", failed.Resource, failed.Err)) + } + } + } + for _, module := range failed.Modules { + if _, named := out[module]; named { + why := skipped[module] + if why == "" { + why = "a step it declares did not complete" + } + reasons[module] = append(reasons[module], "gated: "+why) + } + } + } + for module, why := range skipped { + if _, named := out[module]; named && len(reasons[module]) == 0 { + reasons[module] = append(reasons[module], "gated: "+why) + } + } + for module, why := range reasons { + sort.Strings(why) + out[module] = ModuleResult{Failed: true, Detail: strings.Join(why, "; ")} + } + return out +} diff --git a/internal/apply/module_results_test.go b/internal/apply/module_results_test.go new file mode 100644 index 0000000..2e232be --- /dev/null +++ b/internal/apply/module_results_test.go @@ -0,0 +1,76 @@ +package apply + +import ( + "errors" + "strings" + "testing" + + "github.com/novox/mesh-host/internal/declaration" +) + +func twoModules(t *testing.T) *declaration.Declaration { + t.Helper() + d, err := declaration.Parse([]byte(`{"declaration":1,"builds":{"a":"b-1","b":"b-2"},"resources":[ + {"id":"a.dir","type":"directory","path":"/srv/a","module":"a"}, + {"id":"b.dir","type":"directory","path":"/srv/b","module":"b"}, + {"id":"b.conf","type":"file","path":"/srv/b/c","content":"x","module":"b"}, + {"id":"c.dir","type":"directory","path":"/srv/c","module":"c"}]}`)) + if err != nil { + t.Fatal(err) + } + return d +} + +// 1132/AC3: every resource of module A applied and one of module B failed: A is a success, B failed with the failing +// resource and its error named; a module gated by another's failure is failed too. +func TestAnApplyIsGroupedPerModule(t *testing.T) { + d := twoModules(t) + done := Report{Outcomes: []Outcome{ + {ID: "a.dir", Action: "created"}, + {ID: "c.dir", Action: "skipped", Detail: "a step this module declares did not complete"}, + }} + err := &Error{Resource: "b.dir", Err: errors.New("no space left"), Done: done, Others: 1, + Failures: []Failure{{Resource: "b.dir", Err: errors.New("no space left")}, {Resource: "b.conf", Err: errors.New("denied")}}, + Modules: []string{"c"}} + got := ModuleResults(d, Report{}, err) + if len(got) != 3 { + t.Fatalf("modules %v, want a, b and c", got) + } + if got["a"].Failed || got["a"].Detail != "" { + t.Errorf("a: %+v, want a clean success", got["a"]) + } + b := got["b"] + if !b.Failed || !strings.Contains(b.Detail, "b.dir") || !strings.Contains(b.Detail, "no space left") || + !strings.Contains(b.Detail, "b.conf") || !strings.Contains(b.Detail, "denied") { + t.Errorf("b: %+v, want both failing resources and their errors", b) + } + c := got["c"] + if !c.Failed || !strings.Contains(c.Detail, "did not complete") { + t.Errorf("c: %+v, want it failed as gated, with why", c) + } +} + +// 1132/AC3: an apply that did not fail is a success for every module the declaration names. +func TestACleanApplyIsASuccessForEveryModule(t *testing.T) { + got := ModuleResults(twoModules(t), Report{}, nil) + for _, m := range []string{"a", "b", "c"} { + if r, ok := got[m]; !ok || r.Failed { + t.Errorf("%s: %+v (%v), want a success", m, r, ok) + } + } +} + +// 1132/AC3: a step that belongs to no module stops the apply there; a module with a resource not attempted is not a +// success. +func TestAModuleNotAttemptedIsNotASuccess(t *testing.T) { + d := twoModules(t) + err := &Error{Resource: "boot", Err: errors.New("boom"), Stopped: true, Failures: []Failure{{Resource: "boot", Err: errors.New("boom")}}, + Done: Report{Outcomes: []Outcome{{ID: "a.dir", Action: "created"}}}} + got := ModuleResults(d, Report{}, err) + if got["a"].Failed { + t.Errorf("a was done: %+v", got["a"]) + } + if !got["b"].Failed || !strings.Contains(got["b"].Detail, "boot") { + t.Errorf("b was not attempted: %+v", got["b"]) + } +} diff --git a/internal/apply/move_out.go b/internal/apply/move_out.go index a05d9b8..0deaf35 100644 --- a/internal/apply/move_out.go +++ b/internal/apply/move_out.go @@ -350,7 +350,7 @@ func moveOutPass(d *declaration.Declaration, known store.State, applied map[stri if unknown != declaration.UnknownMoveOut || !applied[r.Identity()] { continue } - module, _ := moduleOf(r.Identity()) + module, _ := d.ModuleOfID(r.Identity()) if module == "" { module = r.Identity() } diff --git a/internal/declaration/declaration.go b/internal/declaration/declaration.go index 0475d6a..1fd3c5b 100644 --- a/internal/declaration/declaration.go +++ b/internal/declaration/declaration.go @@ -103,6 +103,8 @@ type Resource interface { // and not a hash of the content: it is what lets the store say *this is the same resource // I applied last time*, which is what makes removal possible at all. Identity() string + // DeclaredModule is the module the controller says the resource belongs to, or "" (issue 1132). + DeclaredModule() string // Kind is the resource's type, for the store and for reporting. Kind() Type // Target is what the resource acts on, for a person reading a report. @@ -115,12 +117,30 @@ type Resource interface { // strict decoder would otherwise refuse. `Kind()` returns the constant and is what anything // else should read. +// DeclaredModule is on every kind: the module the controller says the resource belongs to, or "" when it does not say +// (an older controller). Read through Declaration.ModuleOfID, which falls back to the guess. +func (d *Directory) DeclaredModule() string { return d.Module } +func (f *File) DeclaredModule() string { return f.Module } +func (u *User) DeclaredModule() string { return u.Module } +func (n *Network) DeclaredModule() string { return n.Module } +func (a *Access) DeclaredModule() string { return a.Module } +func (a *Archive) DeclaredModule() string { return a.Module } +func (d *Process) DeclaredModule() string { return d.Module } +func (s *Service) DeclaredModule() string { return s.Module } +func (o *Opening) DeclaredModule() string { return o.Module } +func (p *Package) DeclaredModule() string { return p.Module } +func (c *Container) DeclaredModule() string { return c.Module } +func (a *Action) DeclaredModule() string { return a.Module } + // Directory is a directory that should exist, with a mode. type Directory struct { ID string `json:"id"` Type Type `json:"type"` - Path string `json:"path"` - Mode string `json:"mode,omitempty"` + // Module is the module this resource belongs to, as the controller says it (novox/hq ADR 0429 §4, issue 1132). Absent + // is an older controller, and the module is then guessed from the id (ModuleGuess). + Module string `json:"module,omitempty"` + Path string `json:"path"` + Mode string `json:"mode,omitempty"` // Owner is the user this belongs to, by name. Absent means root, which is what everything // managed was until users existed. Owner string `json:"owner,omitempty"` @@ -201,8 +221,11 @@ func (d *Directory) validate(where string, _ bool) []string { // File is a file with literal content. The host renders nothing. type File struct { - ID string `json:"id"` - Type Type `json:"type"` + ID string `json:"id"` + Type Type `json:"type"` + // Module is the module this resource belongs to, as the controller says it (novox/hq ADR 0429 §4, issue 1132). Absent + // is an older controller, and the module is then guessed from the id (ModuleGuess). + Module string `json:"module,omitempty"` Path string `json:"path"` Content string `json:"content"` Mode string `json:"mode,omitempty"` @@ -520,7 +543,10 @@ func (f *File) checkOwnership(where string) []string { type User struct { ID string `json:"id"` Type Type `json:"type"` - Name string `json:"name"` + // Module is the module this resource belongs to, as the controller says it (novox/hq ADR 0429 §4, issue 1132). Absent + // is an older controller, and the module is then guessed from the id (ModuleGuess). + Module string `json:"module,omitempty"` + Name string `json:"name"` // Shell this user logs in with. Absent means the host asserts nothing and leaves whatever is // there — the same rule Service.Boot follows, for the same reason: a field that always @@ -571,7 +597,10 @@ const RootNever = "never" type Network struct { ID string `json:"id"` Type Type `json:"type"` - Name string `json:"name"` + // Module is the module this resource belongs to, as the controller says it (novox/hq ADR 0429 §4, issue 1132). Absent + // is an older controller, and the module is then guessed from the id (ModuleGuess). + Module string `json:"module,omitempty"` + Name string `json:"name"` } func (n *Network) Identity() string { return n.ID } @@ -616,7 +645,10 @@ const ( type Access struct { ID string `json:"id"` Type Type `json:"type"` - Path string `json:"path"` + // Module is the module this resource belongs to, as the controller says it (novox/hq ADR 0429 §4, issue 1132). Absent + // is an older controller, and the module is then guessed from the id (ModuleGuess). + Module string `json:"module,omitempty"` + Path string `json:"path"` // Mode is how this module reaches the path: read or read-write. Absent narrows to read. Mode string `json:"mode,omitempty"` @@ -743,6 +775,9 @@ func (u *User) validate(where string, _ bool) []string { type Archive struct { ID string `json:"id"` Type Type `json:"type"` + // Module is the module this resource belongs to, as the controller says it (novox/hq ADR 0429 §4, issue 1132). Absent + // is an older controller, and the module is then guessed from the id (ModuleGuess). + Module string `json:"module,omitempty"` // Source is where to fetch it from. Source string `json:"source"` // Digest is sha256 of the archive, as "sha256:". @@ -796,6 +831,9 @@ func (a *Archive) validate(where string, _ bool) []string { type Process struct { ID string `json:"id"` Type Type `json:"type"` + // Module is the module this resource belongs to, as the controller says it (novox/hq ADR 0429 §4, issue 1132). Absent + // is an older controller, and the module is then guessed from the id (ModuleGuess). + Module string `json:"module,omitempty"` // Name is what the unit is called, and what an operator will see in the process table. Name string `json:"name"` // Source is where to fetch the bundle from, and Digest is what it must hash to. The same @@ -985,7 +1023,10 @@ func (d *Process) validate(where string, _ bool) []string { type Service struct { ID string `json:"id"` Type Type `json:"type"` - Unit string `json:"unit"` + // Module is the module this resource belongs to, as the controller says it (novox/hq ADR 0429 §4, issue 1132). Absent + // is an older controller, and the module is then guessed from the id (ModuleGuess). + Module string `json:"module,omitempty"` + Unit string `json:"unit"` // State is "running" or "stopped" — or absent, and then **the unit's lifecycle is the // machine's; the mesh only reflects its triggers** (novox/hq ADR 0117). The uplink modules // declare the machine's own network manager this way: the mesh writes into its configuration @@ -1157,8 +1198,11 @@ func BlockMarkers(id string) (begin, end string) { // firewall sees a published port after the runtime has translated it, so a forwarded opening names // the container's own port in To as well as the machine's in Port. type Opening struct { - ID string `json:"id"` - Type Type `json:"type"` + ID string `json:"id"` + Type Type `json:"type"` + // Module is the module this resource belongs to, as the controller says it (novox/hq ADR 0429 §4, issue 1132). Absent + // is an older controller, and the module is then guessed from the id (ModuleGuess). + Module string `json:"module,omitempty"` Port int `json:"port"` Protocol string `json:"protocol"` From string `json:"from"` @@ -1223,8 +1267,11 @@ func (o *Opening) validate(where string, _ bool) []string { // Present is the whole of what it asserts, never a version: version is the package manager's // business and the mesh does not hold a second opinion about it. type Package struct { - ID string `json:"id"` - Type Type `json:"type"` + ID string `json:"id"` + Type Type `json:"type"` + // Module is the module this resource belongs to, as the controller says it (novox/hq ADR 0429 §4, issue 1132). Absent + // is an older controller, and the module is then guessed from the id (ModuleGuess). + Module string `json:"module,omitempty"` Package string `json:"package"` // Absent declares that the package is NOT installed (novox/hq ADR 0180): the host removes it // when it is, and leaves a machine that never had it alone. For the one case a module replaces @@ -1249,7 +1296,10 @@ func (p *Package) validate(where string, _ bool) []string { type Container struct { ID string `json:"id"` Type Type `json:"type"` - Name string `json:"name"` + // Module is the module this resource belongs to, as the controller says it (novox/hq ADR 0429 §4, issue 1132). Absent + // is an older controller, and the module is then guessed from the id (ModuleGuess). + Module string `json:"module,omitempty"` + Name string `json:"name"` // Image is pinned by digest (novox/hq ADR 0006) — a tag moves and a digest does not. Image string `json:"image"` Env map[string]string `json:"env,omitempty"` @@ -1555,8 +1605,11 @@ func (c *Container) validate(where string, _ bool) []string { // Action runs something the bundle declared, and the host never learns what it means. type Action struct { - ID string `json:"id"` - Type Type `json:"type"` + ID string `json:"id"` + Type Type `json:"type"` + // Module is the module this resource belongs to, as the controller says it (novox/hq ADR 0429 §4, issue 1132). Absent + // is an older controller, and the module is then guessed from the id (ModuleGuess). + Module string `json:"module,omitempty"` Command []string `json:"command"` // Verify is not optional and is not a courtesy. It is the read-back AND the idempotency // check: the host does not know what a database is, so "is it already there" is a question @@ -1727,6 +1780,58 @@ type Declaration struct { // the only account whose write of a filled file the node-engine's local write socket accepts. Absent is a // controller that names none, and every such write is refused. Provisioner string + + // Builds is the build id of every module this declaration names, by module (novox/hq ADR 0429 §4, issue 1132). + Builds map[string]string + + // declared is what each resource says of its module, by id; nil on a declaration built by hand. + declared map[string]string +} + +// ModuleGuess is the module a resource id names when the declaration does not say: everything before its last dot. +// +// The mesh composes a module's resource ids as `.`, and **a module's name may contain a dot** +// (`novox.be` is one on this mesh) while a resource's own id never does, so the owner is everything before the *last* +// dot. False for what the mesh declares in its own right: the foundation's resources carry no dot, and the +// adoption's are named for the mesh. This is the guess an older controller's declaration is read by; a declaration +// that says `module` is read by ModuleOfID, which never guesses for a resource that says (issue 1132). +func ModuleGuess(id string) (string, bool) { + if strings.HasPrefix(id, AdoptionPrefix) { + return "", false + } + at := strings.LastIndex(id, ".") + if at <= 0 { + return "", false + } + return id[:at], true +} + +// ModuleOfID is the module a resource of this declaration belongs to: the module it says (novox/hq ADR 0429 §4, +// issue 1132), and otherwise the guess from its id (ModuleGuess), so a declaration from an older controller reads as +// it always did. False for what belongs to no module. +func (d *Declaration) ModuleOfID(id string) (string, bool) { + if d != nil { + if d.declared == nil { + for _, r := range d.Resources { + if r.Identity() == id && r.DeclaredModule() != "" { + return r.DeclaredModule(), true + } + } + } else if m, ok := d.declared[id]; ok { + return m, true + } + } + return ModuleGuess(id) +} + +// indexModules keeps what each resource says of its module, so ModuleOfID is a lookup. +func (d *Declaration) indexModules() { + d.declared = map[string]string{} + for _, r := range d.Resources { + if m := r.DeclaredModule(); m != "" { + d.declared[r.Identity()] = m + } + } } // checkProvisioner refuses a provisioner client's account that is no account name, or is root: the account whose @@ -2126,6 +2231,10 @@ type envelope struct { // Provisioner is optional too (novox/hq ADR 0349 §4). **An older host refuses this key**; a controller sends it // only to a host whose health says link.ProvisionerContract. Provisioner string `json:"provisioner_account,omitempty"` + // Builds is optional too (novox/hq ADR 0429 §4, issue 1132): the build id of each module the declaration names, and + // every resource's `module`. **An older host refuses these keys**; a controller sends them only to a host whose + // reports say `reads_modules`. + Builds map[string]string `json:"builds,omitempty"` } func parse(raw []byte, allowActions bool) (*Declaration, error) { @@ -2145,7 +2254,7 @@ func parse(raw []byte, allowActions bool) (*Declaration, error) { d := &Declaration{Version: env.Version, For: env.For, Adoption: env.Adoption, Sequence: env.Sequence, Epoch: env.Epoch, LeftOut: env.LeftOut, Generation: env.Generation, PutBack: env.PutBack, Identities: env.Identities, Groups: env.Groups, Libraries: env.Libraries, Filled: env.Filled, - Provisioner: env.Provisioner} + Provisioner: env.Provisioner, Builds: env.Builds} problems := checkIdentities(env.Identities, env.Groups) problems = append(problems, checkProvisioner(env.Provisioner)...) problems = append(problems, checkLibraries(env.Groups, env.Libraries)...) @@ -2264,6 +2373,7 @@ func parse(raw []byte, allowActions bool) (*Declaration, error) { if len(problems) > 0 { return nil, &RefusalError{Problems: problems} } + d.indexModules() return d, nil } diff --git a/internal/declaration/module_test.go b/internal/declaration/module_test.go new file mode 100644 index 0000000..cdd4b02 --- /dev/null +++ b/internal/declaration/module_test.go @@ -0,0 +1,60 @@ +package declaration + +import "testing" + +// 1132/AC1: a resource says its module, and the declaration says each module's build; a resource that does +// not say gets today's guess, so an older controller still works. +func TestAResourceSaysItsModuleAndTheDeclarationItsBuilds(t *testing.T) { + d, err := Parse([]byte(`{"declaration":1, + "builds":{"novox.be":"b-9","web":"b-4"}, + "resources":[ + {"id":"novox.be.data","type":"directory","path":"/srv/a","module":"novox.be"}, + {"id":"web.conf","type":"file","path":"/etc/w","content":"x","module":"web"}, + {"id":"old.dir","type":"directory","path":"/srv/old"}, + {"id":"base","type":"directory","path":"/srv/base"}]}`)) + if err != nil { + t.Fatal(err) + } + for id, want := range map[string]string{"novox.be.data": "novox.be", "web.conf": "web", "old.dir": "old"} { + got, ok := d.ModuleOfID(id) + if !ok || got != want { + t.Errorf("%s belongs to %q (%v), want %q", id, got, ok, want) + } + } + if m, ok := d.ModuleOfID("base"); ok { + t.Errorf("a resource with no module and no dot belongs to %q, want none", m) + } + if d.Builds["web"] != "b-4" || d.Builds["novox.be"] != "b-9" || len(d.Builds) != 2 { + t.Errorf("builds read as %v", d.Builds) + } +} + +// 1132/AC1: the module the declaration gives wins over the guess, even when the id would guess another. +func TestTheDeclaredModuleWinsOverTheGuess(t *testing.T) { + d, err := Parse([]byte(`{"declaration":1,"resources":[ + {"id":"a.b.c","type":"directory","path":"/srv/a","module":"a.b"}, + {"id":"x.y","type":"directory","path":"/srv/x","module":"other"}]}`)) + if err != nil { + t.Fatal(err) + } + if m, _ := d.ModuleOfID("x.y"); m != "other" { + t.Errorf("x.y belongs to %q, want the declared %q", m, "other") + } +} + +// 1132/AC1: every kind accepts the field. +func TestEveryKindAcceptsAModule(t *testing.T) { + for _, typ := range []Type{TypeDirectory, TypeFile, TypeUser, TypeNetwork, TypeAccess, TypeArchive, TypeProcess, + TypeService, TypeOpening, TypePackage, TypeContainer, TypeAction} { + r := newOf(typ) + if r == nil { + t.Fatalf("no kind %s", typ) + } + if _, ok := r.(interface{ DeclaredModule() string }); !ok { + t.Errorf("%s does not say its module", typ) + } + if unknown := unknownFields([]byte(`{"id":"m.x","type":"`+string(typ)+`","module":"m"}`), r); len(unknown) != 0 { + t.Errorf("%s refuses %v", typ, unknown) + } + } +} diff --git a/internal/link/bus.go b/internal/link/bus.go index fe3ffb2..7d44b8e 100644 --- a/internal/link/bus.go +++ b/internal/link/bus.go @@ -2,9 +2,11 @@ package link import ( "context" + "errors" "fmt" "time" + "git.novox.be/novox/mesh-sdk/go/delivery" "github.com/nats-io/nats.go" ) @@ -115,3 +117,46 @@ func (b OverNATS) Alive(ctx context.Context, node string, body []byte) error { defer cancel() return b.Conn.FlushWithContext(flush) } + +// haveBucket binds to the have bucket the controller made (novox/hq ADR 0429 §4, issue 1132). **Only binds**: a node-engine's +// account reaches no part of the JetStream API beyond what it is granted, and the bucket is the controller's to create +// and to configure, so a bucket that is not there is an error said as what it is. +func (b OverNATS) haveBucket() (nats.KeyValue, error) { + kv, err := b.JS.KeyValue(delivery.HaveBucket) + if err != nil { + return nil, fmt.Errorf("the have bucket %s: %w", delivery.HaveBucket, err) + } + return kv, nil +} + +// HaveGet reads a key of the have bucket with its revision. +func (b OverNATS) HaveGet(_ context.Context, key string) ([]byte, uint64, error) { + kv, err := b.haveBucket() + if err != nil { + return nil, 0, err + } + entry, err := kv.Get(key) + if errors.Is(err, nats.ErrKeyNotFound) { + return nil, 0, ErrHaveAbsent + } + if err != nil { + return nil, 0, err + } + return entry.Value(), entry.Revision(), nil +} + +// HaveSet writes a key of the have bucket over the revision read, or creates it when the revision is zero. +func (b OverNATS) HaveSet(_ context.Context, key string, value []byte, revision uint64) error { + kv, err := b.haveBucket() + if err != nil { + return err + } + if revision == 0 { + _, err = kv.Create(key, value) + return err + } + _, err = kv.Update(key, value, revision) + return err +} + +var _ HaveBus = OverNATS{} diff --git a/internal/link/have.go b/internal/link/have.go new file mode 100644 index 0000000..365d189 --- /dev/null +++ b/internal/link/have.go @@ -0,0 +1,219 @@ +package link + +import ( + "context" + "encoding/json" + "errors" + "fmt" + "sort" + "sync" + + "git.novox.be/novox/mesh-sdk/go/delivery" +) + +// ErrHaveAbsent is what HaveGet returns for a key the bucket does not hold. +var ErrHaveAbsent = errors.New("link: no such key") + +// HaveBus is a link that can read and write the node-engine's own keys in the have bucket (novox/hq ADR 0429 §4). Separate +// from Bus, as HealthBus is: a link that cannot is still a link, and the write waits for the next apply or look. +type HaveBus interface { + // HaveGet is a key's value and last revision, or ErrHaveAbsent. + HaveGet(ctx context.Context, key string) (value []byte, revision uint64, err error) + // HaveSet writes a key over the revision read: zero creates it, and a key that moved since is refused. + HaveSet(ctx context.Context, key string, value []byte, revision uint64) error +} + +// haveState is what the node-engine last learned of each module of the declaration it applied, and of each module's health +// as the last look found it — kept so a look that changes only a health can write only that module's have (issue 1132). +type haveState struct { + mu sync.Mutex + modules map[string]ModuleFact + health map[string]delivery.Health + node *delivery.HaveNode +} + +// moduleHealths is the health of each module a look names, from the states of its resources: failed when any is +// unhealthy, waits when any waits (waiting, starting, held), ok when all are healthy, unknown otherwise. +func moduleHealths(h *Health) map[string]delivery.Health { + out := map[string]delivery.Health{} + if h == nil { + return out + } + rank := map[string]int{} + for _, r := range h.Resources { + if r.Module == "" { + continue + } + var this int + switch r.State { + case StateUnhealthy: + this = 3 + case StateWaiting, StateStarting, StateHeld: + this = 2 + case StateHealthy: + this = 0 + default: + this = 1 + } + if was, seen := rank[r.Module]; !seen || this > was { + rank[r.Module] = this + } + } + for module, r := range rank { + out[module] = [...]delivery.Health{delivery.HealthOK, delivery.HealthUnknown, delivery.HealthWaits, delivery.HealthFailed}[r] + } + return out +} + +// nodeFacts is the node's own facts of a report, each as the JSON the report carries it in. +func nodeFacts(r Report) delivery.HaveNode { + n := delivery.HaveNode{V: delivery.Version} + carry := func(into *json.RawMessage, empty bool, v any) { + if empty { + return + } + if raw, err := json.Marshal(v); err == nil { + *into = raw + } + } + carry(&n.Windows, len(r.Windows) == 0, r.Windows) + carry(&n.Busy, len(r.Busy) == 0, r.Busy) + carry(&n.Strays, len(r.Strays) == 0, r.Strays) + carry(&n.Filters, len(r.Filters) == 0, r.Filters) + carry(&n.Profile, len(r.Profile) == 0, r.Profile) + return n +} + +// noteApply takes what an apply's report says of the modules and the node. +func (q *Queue) noteApply(r Report) bool { + if r.Modules == nil { + return false + } + s := &q.haves + s.mu.Lock() + defer s.mu.Unlock() + s.modules = map[string]ModuleFact{} + for _, m := range r.Modules { + s.modules[m.Module] = m + } + if r.Health != nil { + s.health = moduleHealths(r.Health) + } + n := nodeFacts(r) + s.node = &n + return true +} + +// noteHealth takes a look's health, and says whether any module's changed. +func (q *Queue) noteHealth(h Health) bool { + s := &q.haves + s.mu.Lock() + defer s.mu.Unlock() + next := moduleHealths(&h) + changed := false + for module := range s.modules { + want, ok := next[module] + if !ok { + want = delivery.HealthUnknown + } + have, known := s.health[module] + if !known { + have = delivery.HealthUnknown + } + if have != want { + changed = true + } + } + s.health = next + return changed +} + +// sayHave is a line in the node-engine's log, if it has one. +func (q *Queue) sayHave(s string) { + if q.Say != nil { + q.Say(s) + } +} + +// writeHaves writes this node's keys in the have bucket where they differ from what the bucket holds (novox/hq ADR 0429 +// §4, issue 1132): each module's have, and the node's own facts. It reads the key first and writes nothing when the value +// is the same, otherwise writes over the revision it read, and creates the key when there is none. A write the bucket +// refuses is said and left: the next apply or look writes again, and the report was sent before this ran. +func (q *Queue) writeHaves(ctx context.Context, bus Bus) { + hb, ok := bus.(HaveBus) + if !ok || bus == nil { + return + } + s := &q.haves + s.mu.Lock() + defer s.mu.Unlock() + if s.modules == nil && s.node == nil { + return + } + node := q.Membership.Node + write, cancel := context.WithTimeout(context.WithoutCancel(ctx), q.timeout()) + defer cancel() + modules := make([]string, 0, len(s.modules)) + for module := range s.modules { + modules = append(modules, module) + } + sort.Strings(modules) + for _, module := range modules { + f := s.modules[module] + h := delivery.Have{V: delivery.Version, Build: f.Build, Phase: delivery.PhaseRunning, + Health: delivery.HealthUnknown, Detail: f.Detail} + if f.Failed { + h.Phase = delivery.PhaseFailed + } + if got, known := s.health[module]; known { + h.Health = got + } + key, err := delivery.HaveKey(node, module) + if err != nil { + q.sayHave(fmt.Sprintf("module %q has no key in the have bucket: %v", module, err)) + continue + } + raw, err := delivery.EncodeHave(h) + if err != nil { + q.sayHave(fmt.Sprintf("cannot encode %s: %v", key, err)) + continue + } + q.putHave(write, hb, key, raw, func(current []byte) bool { + c, err := delivery.DecodeHave(current) + return err == nil && delivery.SameHave(c, h) + }) + } + if s.node != nil { + key, err := delivery.HaveNodeKey(node) + if err == nil { + var raw []byte + if raw, err = delivery.EncodeHaveNode(*s.node); err == nil { + want := *s.node + q.putHave(write, hb, key, raw, func(current []byte) bool { + c, err := delivery.DecodeHaveNode(current) + return err == nil && delivery.SameHaveNode(c, want) + }) + } + } + if err != nil { + q.sayHave(fmt.Sprintf("cannot write this node's own facts: %v", err)) + } + } +} + +// putHave reads a key and writes it unless same says it already holds the value. +func (q *Queue) putHave(ctx context.Context, hb HaveBus, key string, value []byte, same func([]byte) bool) { + current, revision, err := hb.HaveGet(ctx, key) + switch { + case errors.Is(err, ErrHaveAbsent): + revision = 0 + case err != nil: + q.sayHave(fmt.Sprintf("could not read %s in the have bucket, and wrote nothing there: %v", key, err)) + return + case same(current): + return + } + if err := hb.HaveSet(ctx, key, value, revision); err != nil { + q.sayHave(fmt.Sprintf("could not write %s in the have bucket; the next apply or look writes it again: %v", key, err)) + } +} diff --git a/internal/link/have_nats_integration_test.go b/internal/link/have_nats_integration_test.go new file mode 100644 index 0000000..1ff1f03 --- /dev/null +++ b/internal/link/have_nats_integration_test.go @@ -0,0 +1,104 @@ +//go:build integration + +package link + +import ( + "context" + "encoding/json" + "sync/atomic" + "testing" + "time" + + "git.novox.be/novox/mesh-sdk/go/delivery" + "github.com/nats-io/nats.go" +) + +// 1132/AC8: against a real bus with the bucket node-engine_have, a node-engine that applies a declaration of two +// modules writes ., . and .node, and a second apply with nothing changed adds no revision. +// +// MESH_TEST_NATS=nats://127.0.0.1:4222 go test -tags integration ./internal/link/ -run TestNatsHavesAreWrittenOnceAndOnlyOnChange +// +// The bucket is made here by the test: in the mesh the controller makes it, and the node-engine only binds to it. +func TestNatsHavesAreWrittenOnceAndOnlyOnChange(t *testing.T) { + conn, js := aBus(t) + const node = "haveok" + kv, err := js.KeyValue(delivery.HaveBucket) + if err != nil { + if kv, err = js.CreateKeyValue(&nats.KeyValueConfig{Bucket: delivery.HaveBucket, History: delivery.HaveHistory}); err != nil { + t.Fatal(err) + } + } + for _, name := range []string{"a", "b", "node"} { + _ = kv.Purge(node + "." + name) + } + + m, key := aMember(t) + m.Node = node + var failB atomic.Bool + var applied atomic.Int32 + failB.Store(true) + apply := func(_ context.Context, raw, _ []byte) Report { + applied.Add(1) + var d struct{ Order } + _ = json.Unmarshal(raw, &d) + return Report{Declared: digest(raw), Order: d.Order, Applied: []string{"a.x", "b.x"}, + Modules: []ModuleFact{{Module: "a", Build: "b-1"}, + {Module: "b", Build: "b-2", Failed: failB.Load(), Detail: "b.x failed: no space"}}, + Health: healthOf(map[string]string{"a": StateHealthy, "b": StateHealthy}), + Windows: []Window{{Step: "backup", Holds: []string{"db"}}}} + } + q := aQueue(m, apply, &keptInMemory{}) + q.attach(context.Background(), OverNATS{Conn: conn, JS: js}) + ctx, stop := context.WithCancel(context.Background()) + worker := runWorker(ctx, q) + defer func() { stop(); <-worker }() + + revisions := func() map[string]uint64 { + out := map[string]uint64{} + for _, name := range []string{"a", "b", "node"} { + if e, err := kv.Get(node + "." + name); err == nil { + out[name] = e.Revision() + } + } + return out + } + arrives := func(what string, id string, seq int64) { + want := applied.Load() + 1 + q.Deliver(ordered(t, key, id, 7, seq)) + eventually(t, what, func() bool { return applied.Load() >= want }) + time.Sleep(300 * time.Millisecond) + } + arrives("the first apply", "one", 1) + eventually(t, "the three keys were never written", func() bool { return len(revisions()) == 3 }) + first := revisions() + + a, _ := kv.Get(node + ".a") + haveA, err := delivery.DecodeHave(a.Value()) + if err != nil || haveA.Phase != delivery.PhaseRunning || haveA.Build != "b-1" || haveA.Health != delivery.HealthOK { + t.Fatalf("a holds %+v (%v)", haveA, err) + } + b, _ := kv.Get(node + ".b") + haveB, err := delivery.DecodeHave(b.Value()) + if err != nil || haveB.Phase != delivery.PhaseFailed || haveB.Detail == "" { + t.Fatalf("b holds %+v (%v)", haveB, err) + } + n, _ := kv.Get(node + ".node") + if hn, err := delivery.DecodeHaveNode(n.Value()); err != nil || len(hn.Windows) == 0 { + t.Fatalf("the node key holds %+v (%v)", hn, err) + } + + arrives("the second apply", "two", 2) + time.Sleep(200 * time.Millisecond) + for name, rev := range revisions() { + if rev != first[name] { + t.Errorf("%s moved from revision %d to %d with nothing changed", name, first[name], rev) + } + } + + failB.Store(false) + arrives("the third apply", "three", 3) + eventually(t, "b was never updated", func() bool { return revisions()["b"] > first["b"] }) + if now := revisions(); now["a"] != first["a"] || now["node"] != first["node"] { + t.Errorf("a key that did not change moved: %v from %v", now, first) + } +} diff --git a/internal/link/have_test.go b/internal/link/have_test.go new file mode 100644 index 0000000..96c8fa0 --- /dev/null +++ b/internal/link/have_test.go @@ -0,0 +1,291 @@ +package link + +import ( + "context" + "encoding/json" + "errors" + "strings" + "sync" + "testing" + "time" + + "git.novox.be/novox/mesh-sdk/go/delivery" +) + +// What the node-engine writes beside its report: a module's have, and the node's own facts (novox/hq ADR 0429 §4, +// issue 1132). + +// haveLink is a link with a bucket behind it that keeps each key's revision as the bus does. +type haveLink struct { + *quietLink + hmu sync.Mutex + kv map[string]haveEntry + sets []string // keys written, in order + events []string // "report" and "have ", in the order they happened + failing bool +} + +type haveEntry struct { + value []byte + rev uint64 +} + +func newHaveLink() *haveLink { return &haveLink{quietLink: newQuietLink(), kv: map[string]haveEntry{}} } + +func (l *haveLink) Report(ctx context.Context, node string, body []byte) error { + err := l.quietLink.Report(ctx, node, body) + if err == nil { + l.hmu.Lock() + l.events = append(l.events, "report") + l.hmu.Unlock() + } + return err +} + +func (l *haveLink) Health(context.Context, string, []byte) error { return nil } + +func (l *haveLink) HaveGet(_ context.Context, key string) ([]byte, uint64, error) { + l.hmu.Lock() + defer l.hmu.Unlock() + e, ok := l.kv[key] + if !ok { + return nil, 0, ErrHaveAbsent + } + return e.value, e.rev, nil +} + +func (l *haveLink) HaveSet(_ context.Context, key string, value []byte, revision uint64) error { + l.hmu.Lock() + defer l.hmu.Unlock() + if l.failing { + return errors.New("the bucket refused") + } + e, ok := l.kv[key] + if (revision == 0) == ok || (ok && e.rev != revision) { + return errors.New("wrong last sequence") + } + l.kv[key] = haveEntry{value: value, rev: e.rev + 1} + l.sets = append(l.sets, key) + l.events = append(l.events, "have "+key) + return nil +} + +func (l *haveLink) have(t *testing.T, key string) delivery.Have { + t.Helper() + l.hmu.Lock() + defer l.hmu.Unlock() + h, err := delivery.DecodeHave(l.kv[key].value) + if err != nil { + t.Fatalf("%s holds no have: %v", key, err) + } + return h +} + +func (l *haveLink) setCount() int { + l.hmu.Lock() + defer l.hmu.Unlock() + return len(l.sets) +} + +func (l *haveLink) put(t *testing.T, key string, h delivery.Have) { + t.Helper() + raw, err := delivery.EncodeHave(h) + if err != nil { + t.Fatal(err) + } + l.hmu.Lock() + defer l.hmu.Unlock() + l.kv[key] = haveEntry{value: raw, rev: l.kv[key].rev + 1} +} + +func healthOf(states map[string]string) *Health { + h := &Health{Contract: LivenessContract, At: time.Now().UTC()} + for module, state := range states { + h.Resources = append(h.Resources, ResourceHealth{Module: module, Resource: module + ".server", State: state}) + } + return h +} + +func appliedModules() Report { + return Report{Declared: "d1", Applied: []string{"a.x", "b.x"}, + Modules: []ModuleFact{{Module: "a", Build: "b-1"}, {Module: "b", Build: "b-2", Failed: true, Detail: "b.x failed: no space"}}, + Health: healthOf(map[string]string{"a": StateHealthy, "b": StateUnhealthy})} +} + +func aHaveQueue(l *haveLink) (*Queue, *[]string) { + var said []string + var mu sync.Mutex + q := &Queue{Membership: Membership{Node: "n1"}, Timeout: time.Second, Say: func(s string) { + mu.Lock() + said = append(said, s) + mu.Unlock() + }} + q.attach(context.Background(), l) + return q, &said +} + +// 1132/AC3, AC5: a module that applied is running with its build and a module that failed is failed with the detail; +// the key does not exist yet, so it is created. +func TestAnAppliedModuleIsWrittenRunningAndAFailedOneFailed(t *testing.T) { + l := newHaveLink() + q, _ := aHaveQueue(l) + q.tell(context.Background(), appliedModules(), applied) + a, b := l.have(t, "n1.a"), l.have(t, "n1.b") + if a.Phase != delivery.PhaseRunning || a.Build != "b-1" || a.Health != delivery.HealthOK { + t.Errorf("a: %+v", a) + } + if b.Phase != delivery.PhaseFailed || b.Build != "b-2" || !strings.Contains(b.Detail, "no space") || + b.Health != delivery.HealthFailed { + t.Errorf("b: %+v", b) + } + if l.kv["n1.a"].rev != 1 { + t.Errorf("a was created at revision %d", l.kv["n1.a"].rev) + } +} + +// 1132/AC5: a have equal to the bucket's value is not written; a different one is written over the last revision. +func TestAHaveIsWrittenOnlyWhenItDiffers(t *testing.T) { + l := newHaveLink() + l.put(t, "n1.a", delivery.Have{V: delivery.Version, Build: "b-1", Phase: delivery.PhaseRunning, Health: delivery.HealthOK}) + l.put(t, "n1.b", delivery.Have{V: delivery.Version, Build: "b-0", Phase: delivery.PhaseRunning, Health: delivery.HealthOK}) + q, _ := aHaveQueue(l) + q.tell(context.Background(), appliedModules(), applied) + if l.kv["n1.a"].rev != 1 { + t.Errorf("a equal to the bucket was written again, to revision %d", l.kv["n1.a"].rev) + } + if l.kv["n1.b"].rev != 2 || l.have(t, "n1.b").Phase != delivery.PhaseFailed { + t.Errorf("b was not updated over its last revision: %+v at %d", l.have(t, "n1.b"), l.kv["n1.b"].rev) + } + before := l.setCount() + q.tell(context.Background(), appliedModules(), applied) + if l.setCount() != before { + t.Errorf("an apply that changed nothing wrote %d keys", l.setCount()-before) + } +} + +// 1132/AC4: a look's health reaches the module's have: ok, failed, waits or unknown, and only a change is written. +func TestAModulesHealthFollowsTheLook(t *testing.T) { + l := newHaveLink() + q, _ := aHaveQueue(l) + rep := Report{Declared: "d1", Modules: []ModuleFact{{Module: "a", Build: "b"}, {Module: "b", Build: "b"}, + {Module: "c", Build: "b"}, {Module: "d", Build: "b"}}} + q.tell(context.Background(), rep, applied) + for _, m := range []string{"a", "b", "c", "d"} { + if h := l.have(t, "n1."+m).Health; h != delivery.HealthUnknown { + t.Errorf("%s with no look is %s, want unknown", m, h) + } + } + h := healthOf(map[string]string{"a": StateHealthy, "b": StateUnhealthy, "c": StateStarting}) + h.Resources = append(h.Resources, ResourceHealth{Module: "a", Resource: "a.two", State: StateWaiting}, + ResourceHealth{Module: "b", Resource: "b.two", State: StateHealthy}, ResourceHealth{Module: "d", Resource: "d.x", State: StateUnknown}) + if !q.SayHealth(context.Background(), *h) { + t.Fatal("not said") + } + for m, want := range map[string]delivery.Health{"a": delivery.HealthWaits, "b": delivery.HealthFailed, + "c": delivery.HealthWaits, "d": delivery.HealthUnknown} { + if got := l.have(t, "n1."+m).Health; got != want { + t.Errorf("%s is %s, want %s", m, got, want) + } + } + before := l.setCount() + h2 := *h + q.SayHealth(context.Background(), h2) + if l.setCount() != before { + t.Errorf("the same look wrote %d keys", l.setCount()-before) + } + h2.Resources = []ResourceHealth{{Module: "a", Resource: "a.server", State: StateHealthy}} + q.SayHealth(context.Background(), h2) + if got := l.have(t, "n1.a").Health; got != delivery.HealthOK { + t.Errorf("a after a healthy look is %s", got) + } +} + +// 1132/AC6: the report is sent first and carries no new key; the node's key holds its windows, busy, strays, filters +// and profile. +func TestTheReportIsSentFirstAndTheNodeKeyCarriesTheNodesFacts(t *testing.T) { + l := newHaveLink() + q, _ := aHaveQueue(l) + rep := appliedModules() + rep.Windows = []Window{{Step: "backup", Holds: []string{"db"}}} + rep.Busy = []Busy{{ID: "x", Container: "x", State: "waiting"}} + rep.Strays = []Stray{{Kind: "container", Name: "old"}} + rep.Filters = []Filter{{Where: "inet mesh", Owner: "mesh"}} + rep.Profile = map[string]any{"cores": 8} + q.tell(context.Background(), rep, applied) + if len(l.events) < 2 || l.events[0] != "report" { + t.Fatalf("the report was not first: %v", l.events) + } + raw, _ := json.Marshal(l.said()[0]) + if strings.Contains(string(raw), "Modules") || strings.Contains(string(raw), "\"modules\"") { + t.Errorf("the report carries the module facts: %s", raw) + } + got, err := delivery.DecodeHaveNode(l.kv["n1.node"].value) + if err != nil { + t.Fatal(err) + } + enc := func(v any) json.RawMessage { b, _ := json.Marshal(v); return b } + want := delivery.HaveNode{V: delivery.Version, Windows: enc(rep.Windows), Busy: enc(rep.Busy), Strays: enc(rep.Strays), + Filters: enc(rep.Filters), Profile: enc(rep.Profile)} + if !delivery.SameHaveNode(got, want) { + t.Errorf("the node key holds %+v, want %+v", got, want) + } +} + +// 1132/AC7: a write the bucket refuses is said in the log; the report went out, and the next apply writes again. +func TestAFailedHaveWriteIsSaidAndNeverBlocksTheReport(t *testing.T) { + l := newHaveLink() + l.failing = true + q, said := aHaveQueue(l) + if !q.tell(context.Background(), appliedModules(), applied) { + t.Fatal("the report did not go out") + } + if len(l.said()) != 1 { + t.Fatalf("reports sent: %d", len(l.said())) + } + if !strings.Contains(strings.Join(*said, "\n"), "n1.a") || !strings.Contains(strings.Join(*said, "\n"), "refused") { + t.Errorf("the failure is not in the log: %v", *said) + } + l.hmu.Lock() + l.failing = false + l.hmu.Unlock() + q.tell(context.Background(), appliedModules(), applied) + if l.have(t, "n1.a").Phase != delivery.PhaseRunning { + t.Error("the next apply did not write again") + } +} + +// A link without a bucket, and a report with no module facts (a refusal), write nothing and say nothing wrong. +func TestNoBucketAndNoFactsWriteNothing(t *testing.T) { + plain := newQuietLink() + q := &Queue{Membership: Membership{Node: "n1"}, Timeout: time.Second} + q.attach(context.Background(), plain) + if !q.tell(context.Background(), appliedModules(), applied) { + t.Fatal("the report did not go out") + } + l := newHaveLink() + q2, _ := aHaveQueue(l) + q2.tell(context.Background(), Report{Refused: "no"}, applied) + if l.setCount() != 0 { + t.Errorf("a refusal wrote %d keys", l.setCount()) + } +} + +// 1132/AC5: a reconcile whose report is not news still writes the have keys that differ, so a node-engine that +// restarted writes them again; and writes nothing when they are the same. +func TestASilentReconcileStillWritesTheHaves(t *testing.T) { + l := newHaveLink() + q, _ := aHaveQueue(l) + q.Reconcile = func(context.Context) (Report, bool) { return appliedModules(), true } + q.reconcile(context.Background(), false) + if len(l.said()) != 0 { + t.Fatalf("a reconcile that is not news said %d reports", len(l.said())) + } + if l.have(t, "n1.a").Phase != delivery.PhaseRunning { + t.Error("the reconcile wrote no have") + } + before := l.setCount() + q.reconcile(context.Background(), false) + if l.setCount() != before { + t.Errorf("an unchanged reconcile wrote %d keys", l.setCount()-before) + } +} diff --git a/internal/link/hearing_nats.go b/internal/link/hearing_nats.go index d8b945f..ab8979d 100644 --- a/internal/link/hearing_nats.go +++ b/internal/link/hearing_nats.go @@ -342,6 +342,14 @@ func (l *natsLink) Health(ctx context.Context, node string, body []byte) error { return OverNATS{Conn: l.conn, JS: l.js}.Health(ctx, node, body) } +func (l *natsLink) HaveGet(ctx context.Context, key string) ([]byte, uint64, error) { + return OverNATS{Conn: l.conn, JS: l.js}.HaveGet(ctx, key) +} + +func (l *natsLink) HaveSet(ctx context.Context, key string, value []byte, revision uint64) error { + return OverNATS{Conn: l.conn, JS: l.js}.HaveSet(ctx, key, value, revision) +} + // Ask asks a module's tool for a declared tool check (novox/hq ADR 0240, to-be 48 §3). The link the // node-engine holds is this one, not OverNATS: without it here the queue's ToolBus assertion failed on // every live link and each tool check read "no link to the bus is open" while the link was up. @@ -370,6 +378,7 @@ func (l *natsLink) Ask(ctx context.Context, subject string, body []byte) ([]byte var ( _ Bus = (*natsLink)(nil) _ HealthBus = (*natsLink)(nil) + _ HaveBus = (*natsLink)(nil) _ ToolBus = (*natsLink)(nil) _ Asker = (*natsLink)(nil) _ Asked = (*natsLink)(nil) diff --git a/internal/link/messages.go b/internal/link/messages.go index f29659a..b480f75 100644 --- a/internal/link/messages.go +++ b/internal/link/messages.go @@ -134,6 +134,11 @@ type Report struct { // know, whole. A node-engine rolled back says so no longer. ReadsGeneration bool `json:"reads_generation,omitempty"` + // ReadsModules says this node-engine reads a resource's `module` and a declaration's `builds` (novox/hq ADR 0429 + // §4, issue 1132), on every report it makes, beside ReadsGeneration: the controller sends them only to a machine + // that said so, for the same reason. + ReadsModules bool `json:"reads_modules,omitempty"` + // OlderGeneration is set on a report refusing a declaration composed from an older assignment // generation than this machine applied (novox/hq issue 234). The refused declaration is the report's // own Declared and Order, so the controller can name who sent it from its record of sends. @@ -196,6 +201,11 @@ type Report struct { // a network manager switched, reaches the mesh at the next push rather than never. Profile map[string]any `json:"profile,omitempty"` + // Modules is how each module of the declaration this report is about came out (novox/hq ADR 0429 §4, issue 1132). **Not + // on the wire**: the report is sent as it always was, and the node-engine writes these beside it as the module's + // have. Nil on a report that is not an apply's. + Modules []ModuleFact `json:"-"` + // Host is the version of the host that produced this report (novox/hq ADR 0141). // // Without it nothing can say a machine is behind, so "every machine current with its source" @@ -851,3 +861,12 @@ type GenerationRefusal struct { Generation int64 `json:"generation"` Applied int64 `json:"applied"` } + +// ModuleFact is how one module's part of an apply came out: the build the node was sent for it, and whether every +// resource of it applied. Detail names the failing resource and its error (issue 1132). +type ModuleFact struct { + Module string + Build string + Failed bool + Detail string +} diff --git a/internal/link/queue.go b/internal/link/queue.go index 9d064a6..1168aa8 100644 --- a/internal/link/queue.go +++ b/internal/link/queue.go @@ -93,6 +93,9 @@ type Queue struct { saying sync.Mutex counted bool numbers numbered + + // haves is what this node-engine writes beside its reports (novox/hq ADR 0429 §4, issue 1132). + haves haveState } // Reconcile is the reconcile's act, run by the queue's worker. See Queue.Reconcile. @@ -265,6 +268,14 @@ func (q *Queue) reconcile(ctx context.Context, asked bool) { return } if !asked && (q.News == nil || !q.News(report)) { + // Not worth a report, but a reconcile found how each module is now: the have keys say it (issue 1132), so a + // node-engine that restarted writes them again without waiting for a change. + if q.noteApply(report) { + q.mu.Lock() + bus := q.bus + q.mu.Unlock() + q.writeHaves(ctx, bus) + } return } if asked { @@ -324,6 +335,10 @@ func (q *Queue) tell(ctx context.Context, r Report, kind kindOf, settle ...Decla q.unasked = &r } if published { + // After the report, which is sent first and unchanged; a write the bucket refuses never undoes it (issue 1132). + if q.noteApply(r) { + q.writeHaves(ctx, bus) + } if keep && q.Unsaid != nil { if err := q.Unsaid.Said(r.Declared); err != nil { q.Say("said this apply's report, and cannot forget it: " + err.Error()) @@ -354,6 +369,8 @@ func (q *Queue) stamp(r Report) Report { // Every report of a node-engine that orders its reports says it reads a generation (novox/hq issue // 234): this build does, and the controller sends one only to a machine that said so. r.ReadsGeneration = true + // And the module of each resource and the build of each module (novox/hq ADR 0429 §4, issue 1132), the same way. + r.ReadsModules = true return r } @@ -538,6 +555,10 @@ func (q *Queue) SayHealth(ctx context.Context, h Health) bool { q.Say("could not say this machine's health: " + err.Error()) return false } + // A look that changed a module's health changes its have (issue 1132). + if q.noteHealth(h) { + q.writeHaves(ctx, bus) + } return true } diff --git a/internal/link/queue_test.go b/internal/link/queue_test.go index 9443983..3e0e7b2 100644 --- a/internal/link/queue_test.go +++ b/internal/link/queue_test.go @@ -563,3 +563,26 @@ func TestEveryReportSaysItReadsAGeneration(t *testing.T) { t.Fatalf("the report does not say it reads a generation: %s", raw) } } + +// 1132/AC2: every report says this node-engine reads a resource's module and a declaration's builds, beside the +// generation: the controller sends them only to a machine that said so, because an older node-engine refuses the keys. +func TestEveryReportSaysItReadsModules(t *testing.T) { + m, key := aMember(t) + l := newQuietLink() + a := &applies{} + q := aQueue(m, a.apply, &keptInMemory{}) + q.attach(context.Background(), l) + q.Deliver(ordered(t, key, "new", 7, 12)) + ctx, stop := context.WithCancel(context.Background()) + worker := runWorker(ctx, q) + eventually(t, "the delivery was never reported", func() bool { return len(l.said()) == 1 }) + stop() + <-worker + raw, err := json.Marshal(l.said()[0]) + if err != nil { + t.Fatal(err) + } + if !strings.Contains(string(raw), `"reads_modules":true`) || !strings.Contains(string(raw), `"reads_generation":true`) { + t.Fatalf("the report does not say it reads modules beside the generation: %s", raw) + } +} diff --git a/internal/liveness/liveness.go b/internal/liveness/liveness.go index 0a5bc94..51cb868 100644 --- a/internal/liveness/liveness.go +++ b/internal/liveness/liveness.go @@ -100,7 +100,7 @@ func LongRunning(d *declaration.Declaration, held map[string]bool) []Resource { return nil } for _, r := range d.Resources { - module, ok := ModuleOf(r.Identity()) + module, ok := d.ModuleOfID(r.Identity()) if !ok || held[r.Identity()] { continue } @@ -133,16 +133,7 @@ func LongRunning(d *declaration.Declaration, held map[string]bool) []Resource { // ModuleOf is the module a resource's id belongs to: everything before its last dot — a module's name may // carry a dot, its resources' own ids never do (the apply's moduleOf, for the same reason). False for // what the mesh declares in its own right. -func ModuleOf(id string) (string, bool) { - if strings.HasPrefix(id, declaration.AdoptionPrefix) { - return "", false - } - at := strings.LastIndex(id, ".") - if at <= 0 { - return "", false - } - return id[:at], true -} +func ModuleOf(id string) (string, bool) { return declaration.ModuleGuess(id) } // Observed is one resource as one read found it. type Observed struct { -- 2.54.0 From 9d9e3a46c4f5943cfffb413d962ce48b75e2fd39 Mon Sep 17 00:00:00 2001 From: jochen Date: Sun, 18 Oct 2026 19:18:08 +0200 Subject: [PATCH 2/3] Bind the have bucket once per pass, bound by the queue's timeout, and stop the pass at the first error (issue 1132) --- internal/link/bus.go | 52 ++++++++++++++++------------ internal/link/have.go | 55 +++++++++++++++++++++--------- internal/link/have_test.go | 64 +++++++++++++++++++++++++++++++---- internal/link/hearing_nats.go | 8 ++--- 4 files changed, 129 insertions(+), 50 deletions(-) diff --git a/internal/link/bus.go b/internal/link/bus.go index 7d44b8e..c07d8fd 100644 --- a/internal/link/bus.go +++ b/internal/link/bus.go @@ -118,24 +118,38 @@ func (b OverNATS) Alive(ctx context.Context, node string, body []byte) error { return b.Conn.FlushWithContext(flush) } -// haveBucket binds to the have bucket the controller made (novox/hq ADR 0429 §4, issue 1132). **Only binds**: a node-engine's -// account reaches no part of the JetStream API beyond what it is granted, and the bucket is the controller's to create -// and to configure, so a bucket that is not there is an error said as what it is. -func (b OverNATS) haveBucket() (nats.KeyValue, error) { - kv, err := b.JS.KeyValue(delivery.HaveBucket) +// HaveBind binds the have bucket the controller made (novox/hq ADR 0429 §4, issue 1132), once for a pass of writes. +// **Only binds**: a node-engine's account reaches no part of the JetStream API beyond what it is granted, and the bucket +// is the controller's to create and to configure, so a bucket that is not there is an error said as what it is. Every +// call on the bucket waits no longer than ctx has left: a refused publish is answered by nothing, and the client's own +// wait is five seconds. +func (b OverNATS) HaveBind(ctx context.Context) (HaveBucket, error) { + js := b.JS + wait := 5 * time.Second + if at, ok := ctx.Deadline(); ok { + wait = time.Until(at) + if wait <= 0 { + return nil, context.DeadlineExceeded + } + } + if b.Conn != nil { + bounded, err := b.Conn.JetStream(nats.MaxWait(wait)) + if err != nil { + return nil, err + } + js = bounded + } + kv, err := js.KeyValue(delivery.HaveBucket) if err != nil { return nil, fmt.Errorf("the have bucket %s: %w", delivery.HaveBucket, err) } - return kv, nil + return natsHaveBucket{kv}, nil } -// HaveGet reads a key of the have bucket with its revision. -func (b OverNATS) HaveGet(_ context.Context, key string) ([]byte, uint64, error) { - kv, err := b.haveBucket() - if err != nil { - return nil, 0, err - } - entry, err := kv.Get(key) +type natsHaveBucket struct{ kv nats.KeyValue } + +func (n natsHaveBucket) Get(_ context.Context, key string) ([]byte, uint64, error) { + entry, err := n.kv.Get(key) if errors.Is(err, nats.ErrKeyNotFound) { return nil, 0, ErrHaveAbsent } @@ -145,17 +159,13 @@ func (b OverNATS) HaveGet(_ context.Context, key string) ([]byte, uint64, error) return entry.Value(), entry.Revision(), nil } -// HaveSet writes a key of the have bucket over the revision read, or creates it when the revision is zero. -func (b OverNATS) HaveSet(_ context.Context, key string, value []byte, revision uint64) error { - kv, err := b.haveBucket() - if err != nil { - return err - } +func (n natsHaveBucket) Set(_ context.Context, key string, value []byte, revision uint64) error { + var err error if revision == 0 { - _, err = kv.Create(key, value) + _, err = n.kv.Create(key, value) return err } - _, err = kv.Update(key, value, revision) + _, err = n.kv.Update(key, value, revision) return err } diff --git a/internal/link/have.go b/internal/link/have.go index 365d189..99ef52b 100644 --- a/internal/link/have.go +++ b/internal/link/have.go @@ -14,13 +14,20 @@ import ( // ErrHaveAbsent is what HaveGet returns for a key the bucket does not hold. var ErrHaveAbsent = errors.New("link: no such key") -// HaveBus is a link that can read and write the node-engine's own keys in the have bucket (novox/hq ADR 0429 §4). Separate -// from Bus, as HealthBus is: a link that cannot is still a link, and the write waits for the next apply or look. +// HaveBus is a link that can bind the have bucket (novox/hq ADR 0429 §4). Separate from Bus, as HealthBus is: a link that +// cannot is still a link, and the write waits for the next apply or look. type HaveBus interface { - // HaveGet is a key's value and last revision, or ErrHaveAbsent. - HaveGet(ctx context.Context, key string) (value []byte, revision uint64, err error) - // HaveSet writes a key over the revision read: zero creates it, and a key that moved since is refused. - HaveSet(ctx context.Context, key string, value []byte, revision uint64) error + // HaveBind binds the bucket the controller made, once for a pass of writes, within ctx. A node whose grant is not there + // yet gets an error here (or the context's end), and writes nothing. + HaveBind(ctx context.Context) (HaveBucket, error) +} + +// HaveBucket is the bound bucket. +type HaveBucket interface { + // Get is a key's value and last revision, or ErrHaveAbsent. + Get(ctx context.Context, key string) (value []byte, revision uint64, err error) + // Set writes a key over the revision read: zero creates it, and a key that moved since is refused. + Set(ctx context.Context, key string, value []byte, revision uint64) error } // haveState is what the node-engine last learned of each module of the declaration it applied, and of each module's health @@ -151,8 +158,15 @@ func (q *Queue) writeHaves(ctx context.Context, bus Bus) { return } node := q.Membership.Node + // The whole pass is bounded by the queue's timeout, and it stops at the first error: a node whose grant is not there + // yet would otherwise make the worker wait the bound again for every key (issue 1132). write, cancel := context.WithTimeout(context.WithoutCancel(ctx), q.timeout()) defer cancel() + bucket, err := hb.HaveBind(write) + if err != nil { + q.sayHave(fmt.Sprintf("could not bind the have bucket, and wrote nothing there; the next apply or look writes again: %v", err)) + return + } modules := make([]string, 0, len(s.modules)) for module := range s.modules { modules = append(modules, module) @@ -178,10 +192,12 @@ func (q *Queue) writeHaves(ctx context.Context, bus Bus) { q.sayHave(fmt.Sprintf("cannot encode %s: %v", key, err)) continue } - q.putHave(write, hb, key, raw, func(current []byte) bool { + if !q.putHave(write, bucket, key, raw, func(current []byte) bool { c, err := delivery.DecodeHave(current) return err == nil && delivery.SameHave(c, h) - }) + }) { + return + } } if s.node != nil { key, err := delivery.HaveNodeKey(node) @@ -189,7 +205,7 @@ func (q *Queue) writeHaves(ctx context.Context, bus Bus) { var raw []byte if raw, err = delivery.EncodeHaveNode(*s.node); err == nil { want := *s.node - q.putHave(write, hb, key, raw, func(current []byte) bool { + q.putHave(write, bucket, key, raw, func(current []byte) bool { c, err := delivery.DecodeHaveNode(current) return err == nil && delivery.SameHaveNode(c, want) }) @@ -201,19 +217,26 @@ func (q *Queue) writeHaves(ctx context.Context, bus Bus) { } } -// putHave reads a key and writes it unless same says it already holds the value. -func (q *Queue) putHave(ctx context.Context, hb HaveBus, key string, value []byte, same func([]byte) bool) { - current, revision, err := hb.HaveGet(ctx, key) +// putHave reads a key and writes it unless same says it already holds the value. False when the pass should stop: the +// bucket failed, the error is said once, and the rest wait for the next pass. +func (q *Queue) putHave(ctx context.Context, b HaveBucket, key string, value []byte, same func([]byte) bool) bool { + if ctx.Err() != nil { + q.sayHave(fmt.Sprintf("the time for writing the have bucket ran out before %s; the next apply or look writes it", key)) + return false + } + current, revision, err := b.Get(ctx, key) switch { case errors.Is(err, ErrHaveAbsent): revision = 0 case err != nil: - q.sayHave(fmt.Sprintf("could not read %s in the have bucket, and wrote nothing there: %v", key, err)) - return + q.sayHave(fmt.Sprintf("could not read %s in the have bucket, and wrote nothing more there: %v", key, err)) + return false case same(current): - return + return true } - if err := hb.HaveSet(ctx, key, value, revision); err != nil { + if err := b.Set(ctx, key, value, revision); err != nil { q.sayHave(fmt.Sprintf("could not write %s in the have bucket; the next apply or look writes it again: %v", key, err)) + return false } + return true } diff --git a/internal/link/have_test.go b/internal/link/have_test.go index 96c8fa0..90f465b 100644 --- a/internal/link/have_test.go +++ b/internal/link/have_test.go @@ -18,11 +18,14 @@ import ( // haveLink is a link with a bucket behind it that keeps each key's revision as the bus does. type haveLink struct { *quietLink - hmu sync.Mutex - kv map[string]haveEntry - sets []string // keys written, in order - events []string // "report" and "have ", in the order they happened - failing bool + hmu sync.Mutex + kv map[string]haveEntry + sets []string // keys written, in order + events []string // "report" and "have ", in the order they happened + failing bool + binds int + setTries int + bindFails, bindStuck bool } type haveEntry struct { @@ -44,7 +47,22 @@ func (l *haveLink) Report(ctx context.Context, node string, body []byte) error { func (l *haveLink) Health(context.Context, string, []byte) error { return nil } -func (l *haveLink) HaveGet(_ context.Context, key string) ([]byte, uint64, error) { +func (l *haveLink) HaveBind(ctx context.Context) (HaveBucket, error) { + l.hmu.Lock() + l.binds++ + stuck, refuse := l.bindStuck, l.bindFails + l.hmu.Unlock() + if stuck { + <-ctx.Done() + return nil, ctx.Err() + } + if refuse { + return nil, errors.New("no grant for the have bucket") + } + return l, nil +} + +func (l *haveLink) Get(_ context.Context, key string) ([]byte, uint64, error) { l.hmu.Lock() defer l.hmu.Unlock() e, ok := l.kv[key] @@ -54,9 +72,10 @@ func (l *haveLink) HaveGet(_ context.Context, key string) ([]byte, uint64, error return e.value, e.rev, nil } -func (l *haveLink) HaveSet(_ context.Context, key string, value []byte, revision uint64) error { +func (l *haveLink) Set(_ context.Context, key string, value []byte, revision uint64) error { l.hmu.Lock() defer l.hmu.Unlock() + l.setTries++ if l.failing { return errors.New("the bucket refused") } @@ -289,3 +308,34 @@ func TestASilentReconcileStillWritesTheHaves(t *testing.T) { t.Errorf("an unchanged reconcile wrote %d keys", l.setCount()-before) } } + +// 1132/AC7: a bind that fails, or never answers, writes nothing and returns within the queue's timeout; a write that +// fails ends the pass, and is said once. +func TestAFailedBindOrWriteEndsThePassAndReturnsInTime(t *testing.T) { + for name, set := range map[string]func(*haveLink){ + "a bind refused": func(l *haveLink) { l.bindFails = true }, + "a bind that never answers": func(l *haveLink) { l.bindStuck = true }, + } { + l := newHaveLink() + set(l) + q, said := aHaveQueue(l) + q.Timeout = 100 * time.Millisecond + start := time.Now() + if !q.tell(context.Background(), appliedModules(), applied) { + t.Fatalf("%s: the report did not go out", name) + } + if took := time.Since(start); took > time.Second { + t.Errorf("%s: the pass took %s", name, took) + } + if l.binds != 1 || l.setTries != 0 || len(*said) != 1 { + t.Errorf("%s: binds %d, writes tried %d, said %v; want one bind, no write, one line", name, l.binds, l.setTries, *said) + } + } + l := newHaveLink() + l.failing = true + q, said := aHaveQueue(l) + q.tell(context.Background(), appliedModules(), applied) + if l.binds != 1 || l.setTries != 1 || len(*said) != 1 { + t.Errorf("a failed write: binds %d, writes tried %d, said %v; want one of each and one line", l.binds, l.setTries, *said) + } +} diff --git a/internal/link/hearing_nats.go b/internal/link/hearing_nats.go index ab8979d..dbe3d79 100644 --- a/internal/link/hearing_nats.go +++ b/internal/link/hearing_nats.go @@ -342,12 +342,8 @@ func (l *natsLink) Health(ctx context.Context, node string, body []byte) error { return OverNATS{Conn: l.conn, JS: l.js}.Health(ctx, node, body) } -func (l *natsLink) HaveGet(ctx context.Context, key string) ([]byte, uint64, error) { - return OverNATS{Conn: l.conn, JS: l.js}.HaveGet(ctx, key) -} - -func (l *natsLink) HaveSet(ctx context.Context, key string, value []byte, revision uint64) error { - return OverNATS{Conn: l.conn, JS: l.js}.HaveSet(ctx, key, value, revision) +func (l *natsLink) HaveBind(ctx context.Context) (HaveBucket, error) { + return OverNATS{Conn: l.conn, JS: l.js}.HaveBind(ctx) } // Ask asks a module's tool for a declared tool check (novox/hq ADR 0240, to-be 48 §3). The link the -- 2.54.0 From aab65b81f65c3a2984e0c3b8dd7af1219d2650e5 Mon Sep 17 00:00:00 2001 From: jochen Date: Sun, 18 Oct 2026 19:28:07 +0200 Subject: [PATCH 3/3] Read the build agent's manifest from the controller's repository beside the check, where it moved (issue 1151) --- internal/bootstrap/builder_test.go | 41 ++++++++++++++++++++++++++---- 1 file changed, 36 insertions(+), 5 deletions(-) diff --git a/internal/bootstrap/builder_test.go b/internal/bootstrap/builder_test.go index f4158f1..6127d47 100644 --- a/internal/bootstrap/builder_test.go +++ b/internal/bootstrap/builder_test.go @@ -148,21 +148,26 @@ func TestAForgeManifestTheCatalogueDoesNotHaveIsNamed(t *testing.T) { } } -// theCataloguesModule is the catalogue's own manifest of a module, never a copy: the clone beside a merge -// check (MESH_CHECK_BESIDE), where it must be; else a checkout beside this one; else the test skips aloud. +// theCataloguesModule is a module's own manifest, never a copy: the clone beside a merge check (MESH_CHECK_BESIDE), where it +// must be; else a checkout beside this one; else the test skips aloud. The catalogue holds it, but for the build agent, +// which moved whole into the controller's repository (ADR 0420, issue 1019, issue 1151) and is read from there. func theCataloguesModule(t *testing.T, module string) []byte { t.Helper() - inside := filepath.Join("mesh-catalog", "modules", module, "module.json") + repository := "mesh-catalog" + if module == BuilderModule { + repository = "mesh-controller" + } + inside := filepath.Join(repository, "modules", module, "module.json") if root := os.Getenv("MESH_CHECK_BESIDE"); root != "" { raw, err := os.ReadFile(filepath.Join(root, inside)) if err != nil { - t.Fatalf("the catalogue beside this check (MESH_CHECK_BESIDE=%s) has no %s: %v", root, inside, err) + t.Fatalf("%s beside this check (MESH_CHECK_BESIDE=%s) has no %s: %v", repository, root, inside, err) } return raw } raw, err := os.ReadFile(filepath.Join("..", "..", "..", inside)) if err != nil { - t.Skipf("NOT JUDGED: no catalogue beside this checkout and no MESH_CHECK_BESIDE; the merge check judges it") + t.Skipf("NOT JUDGED: no %s beside this checkout and no MESH_CHECK_BESIDE; the merge check judges it", repository) } return raw } @@ -217,3 +222,29 @@ func TestABuildAgentManifestWithNowhereForTheCarriedBuilderIsRefused(t *testing. t.Fatalf("a manifest with no image artifact was pinned: %v", err) } } + +// 1151: the build agent's manifest lives in the controller's repository now (ADR 0420, issue 1019), and a merge check of +// this repository has the controller cloned beside it, so a tree shaped like that check's, whose catalogue no longer holds +// the build agent, is read from the controller. +func TestTheBuildAgentsManifestIsReadFromTheControllerBesideTheCheck(t *testing.T) { + beside := t.TempDir() + for path, body := range map[string]string{ + "mesh-catalog/modules/nats/module.json": `{"module":"nats"}`, + "mesh-controller/modules/build-agent/module.json": `{"module":"build-agent"}`, + } { + full := filepath.Join(beside, path) + if err := os.MkdirAll(filepath.Dir(full), 0o755); err != nil { + t.Fatal(err) + } + if err := os.WriteFile(full, []byte(body), 0o644); err != nil { + t.Fatal(err) + } + } + t.Setenv("MESH_CHECK_BESIDE", beside) + if got := string(theCataloguesModule(t, BuilderModule)); got != `{"module":"build-agent"}` { + t.Errorf("read %s", got) + } + if got := string(theCataloguesModule(t, "nats")); got != `{"module":"nats"}` { + t.Errorf("a module the catalogue holds was read as %s", got) + } +} -- 2.54.0