novox/mesh-catalog
9
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Merge branch 'feat/47-G-recover-readers' into feat/47-G-join8

Browse Source 
# Conflicts:
#	modules/mesh-config/module.json
#	modules/mesh-provisioner/module.json
#	modules/mesh-self-check/module.json
This commit is contained in:
jschoubben
2026-11-12 00:14:26 +01:00
parent ebaa4c9946 e0428bd874
commit 4b749286bd
43 changed files with 1256 additions and 463 deletions
Download Patch File Download Diff File Expand all files Collapse all files
modules/claude-code/cmd/claude-code/catalogfeed.go
+60
View File
@@ -0,0 +1,60 @@
package main
// claude-code's reading of which modules exist and of each placed module's manifest (design 47 G §196, §594, §614): the
// catalog half, taken through meshcatalogclient's follower of the watched branches on the part's one views reader. A
// module's manifest is the one its catalog graph entry carries whole (§443), never the indexer's state `modules` or
// `manifests`.
import (
"fmt"
"strings"
"git.novox.be/novox/mesh-delivery/contracts/go/clients/meshcatalogclient"
catalog "git.novox.be/novox/mesh-delivery/contracts/go/seats/meshcatalog"
)
// CatalogHalf is what this module reads of the catalog half.
type CatalogHalf interface {
// Entry is a module's entry at the trunk; an error while a watched branch has no first view yet, never "no module".
Entry(module string) (catalog.CatalogEntry, bool, error)
// Ready is nil once every watched branch has its first view (none-yet counting as one), else says what it waits for.
Ready() error
}
// branchesHalf is the CatalogHalf over the follower of the watched branches.
type branchesHalf struct{ b *meshcatalogclient.Branches }
func (h branchesHalf) Ready() error {
followed := h.b.Followed()
if len(followed) == 0 {
return fmt.Errorf("the catalog half names no watched branch yet (the seat mesh-catalog's registrations are asked until they answer)")
}
var waiting []string
for _, branch := range followed {
if c, ok := h.b.Catalog(branch); ok {
select {
case <-c.Ready():
default:
waiting = append(waiting, branch)
}
}
}
if len(waiting) > 0 {
return fmt.Errorf("the catalog half has no first view of %s yet", strings.Join(waiting, ", "))
}
return nil
}
func (h branchesHalf) Entry(module string) (catalog.CatalogEntry, bool, error) {
if err := h.Ready(); err != nil {
return catalog.CatalogEntry{}, false, err
}
for _, branch := range h.b.Followed() { // the branch followed first names a module more than one hold
if c, ok := h.b.Catalog(branch); ok {
if e, ok := c.Entry(module); ok {
return e, true, nil
}
}
}
return catalog.CatalogEntry{}, false, nil
}
modules/claude-code/cmd/claude-code/catalogfeed_test.go
+90
View File
@@ -0,0 +1,90 @@
package main
import (
"context"
"encoding/json"
"fmt"
"strings"
"sync"
"testing"
"time"
"git.novox.be/novox/mesh-delivery/contracts/go/clients/meshcatalogclient"
catalog "git.novox.be/novox/mesh-delivery/contracts/go/seats/meshcatalog"
delivery "git.novox.be/novox/mesh-delivery/contracts/go/seats/meshdelivery"
"git.novox.be/novox/mesh-sdk/go/views"
)
// wiringOf is a part's reader in miniature: it keeps the followers adopted, so a test starts them as the reader's Start does.
type wiringOf struct {
mu sync.Mutex
fs []*views.Follower
}
func (w *wiringOf) Adopt(f *views.Follower, _, _ string, _ func(context.Context, []byte) error) error {
w.mu.Lock()
defer w.mu.Unlock()
w.fs = append(w.fs, f)
return nil
}
func (w *wiringOf) Drop(string, string) error { return nil }
// Design 47 G §196, §594: which modules exist, and the manifest of each, are the catalog half's; a branch with no first
// view yet is an error naming it, never no module, and an entry's manifest is the one it carries whole.
func TestCatalogHalfIsTheWatchedBranchesNewestRevision(t *testing.T) {
ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
defer cancel()
answered := false
var mu sync.Mutex
call := func(key string, _ any) (json.RawMessage, error) {
mu.Lock()
defer mu.Unlock()
switch key {
case meshcatalogclient.RegistrationsVerb:
return json.Marshal(catalog.RegistrationsAnswer{Registrations: []catalog.Registration{{Module: "m", Version: 1, Branches: []string{"main"}}}})
case meshcatalogclient.SourceVerb:
return json.Marshal(delivery.SourceAnswer{})
case meshcatalogclient.RevisionVerb:
if !answered {
return nil, fmt.Errorf("the holder is away")
}
return json.Marshal(catalog.RevisionAnswer{Revision: &catalog.CatalogRevision{Revision: 1, Kind: meshcatalogclient.KindTrunk},
Entries: []catalog.CatalogEntry{{Module: "pg", Manifest: map[string]any{"module": "pg"}}}})
}
return nil, fmt.Errorf("asked %s", key)
}
w := &wiringOf{}
b, err := meshcatalogclient.NewBranches(w, call)
if err != nil {
t.Fatal(err)
}
h := branchesHalf{b}
if _, _, err := h.Entry("pg"); err == nil || !strings.Contains(err.Error(), "no watched branch") {
t.Fatalf("before the registrations: %v, want an error naming the missing branch", err)
}
go func() { _ = w.fs[0].Start(ctx) }() // the registrations
for len(b.Followed()) == 0 {
select {
case <-ctx.Done():
t.Fatal("the registrations' branch is never followed")
case <-time.After(10 * time.Millisecond):
}
}
if _, _, err := h.Entry("pg"); err == nil || !strings.Contains(err.Error(), "no first view of main") {
t.Fatalf("with no first view: %v, want an error naming the branch, never no module", err)
}
mu.Lock()
answered = true
mu.Unlock()
c, _ := b.Catalog("main")
if err := c.Start(ctx); err != nil {
t.Fatal(err)
}
got, ok, err := h.Entry("pg")
if err != nil || !ok || got.Manifest["module"] != "pg" || h.Ready() != nil {
t.Fatalf("with the first view: %v, %v, %v", got, ok, err)
}
if _, ok, err := h.Entry("absent"); ok || err != nil {
t.Fatalf("a module the half does not list: %v, %v", ok, err)
}
}
modules/claude-code/cmd/claude-code/config_test.go
+2 -2
View File
@@ -277,9 +277,9 @@ func TestTheManifestListsEveryToolServed(t *testing.T) {
_ = json.Unmarshal(raw, &m)
// The hand-act seat's warranted and hand-act, the seat mesh-catalog's seat list (ADR 0635 §1), and the owners' read tools of the three followed states; no verb of
// the controller, and no read of the indexer's or the assigner's state (ADR 0615 §1): their views are consumed.
if !slices.Equal(m.Invokes, []string{"seat:mesh-hand-acts.warranted", "seat:mesh-hand-acts.hand-act", "seat:mesh-catalog.seat-definitions", ModulesTool, ManifestsTool, AssignmentsTool}) ||
if !slices.Equal(m.Invokes, []string{"seat:mesh-hand-acts.warranted", "seat:mesh-hand-acts.hand-act", "seat:mesh-catalog.seat-definitions", RevisionVerb, RegistrationsVerb, SourceVerb, AssignmentsTool}) ||
!slices.Equal(m.Reads, []string{"claude-licence-manager.bindings"}) ||
!slices.Equal(m.Consumes, []string{"mesh-delivery-indexer.views.modules", "mesh-delivery-indexer.views.manifests", "mesh-delivery-assigner.views.assignments", "mesh-catalog.seat-definition-changed", "mesh-catalog.seat-definition-removed", "mesh-conditions.findings-asked"}) {
!slices.Equal(m.Consumes, []string{"mesh-catalog.catalog-revision", "mesh-catalog.registration-changed", "mesh-delivery.source-moved", "mesh-delivery-assigner.views.assignments", "mesh-catalog.seat-definition-changed", "mesh-catalog.seat-definition-removed", "mesh-conditions.findings-asked"}) {
t.Errorf("invokes = %v, reads = %v, consumes = %v", m.Invokes, m.Reads, m.Consumes)
}
if src, _ := os.ReadFile("proposals.go"); !strings.Contains(string(src), `"seat:mesh-hand-acts.warranted"`) ||
modules/claude-code/cmd/claude-code/insteadof_test.go
+36 -23
View File
@@ -2,13 +2,10 @@ package main
import (
"context"
"crypto/sha256"
"encoding/json"
"errors"
"fmt"
assignerclient "git.novox.be/novox/mesh-contracts/go/clients/meshmoduleassignerclient"
mesh "git.novox.be/novox/mesh-contracts/go/seats/meshmoduleassigner"
catalogclient "git.novox.be/novox/mesh-delivery/contracts/go/clients/meshcatalogclient"
catalog "git.novox.be/novox/mesh-delivery/contracts/go/seats/meshcatalog"
"os"
"reflect"
@@ -57,7 +54,7 @@ func someSeats() []seatdef.Definition {
// saying what its tool replaces, as the seat mesh-module-assigner and the indexer keep them on the bus.
func aMesh(t *testing.T) aBus {
t.Helper()
b := aBus{AssignmentsRead: {}, ModulesRead: {}, ManifestsRead: {}}
b := aBus{AssignmentsRead: {}}
for _, p := range []mesh.Assignments{
{V: 1, Node: "anchor", Decided: mesh.NodeDecided{PublicDomain: "Anchor.Example"},
Modules: []mesh.PlacedModule{{Module: "docker"}, {Module: "zsh"}}},
@@ -71,21 +68,33 @@ func aMesh(t *testing.T) aBus {
b[AssignmentsRead][k] = v
}
}
return b
}
// fakeCatalog is the catalog half as the tests give it: docker and zsh, docker's manifest saying what its tool replaces.
type fakeCatalog struct {
entries map[string]catalog.CatalogEntry
notYet error
}
func (f fakeCatalog) Entry(module string) (catalog.CatalogEntry, bool, error) {
e, ok := f.entries[module]
return e, ok, f.notYet
}
func (f fakeCatalog) Ready() error { return f.notYet }
func aCatalog() fakeCatalog {
f := fakeCatalog{entries: map[string]catalog.CatalogEntry{}}
for module, manifest := range map[string]string{
"docker": `{"module":"docker","replaces":{"docker_logs":["docker logs"]}}`,
"zsh": `{"module":"zsh"}`,
} {
digest := fmt.Sprintf("%x", sha256.Sum256([]byte(manifest)))
mkey, _ := catalogclient.ManifestsKey(digest)
raw, err := json.Marshal(catalog.ManifestsByDigestValue{V: 1, Module: module, Manifest: manifest})
if err != nil {
t.Fatal(err)
}
b[ManifestsRead][mkey] = raw
key, _ := catalogclient.ModulesKey(module)
b[ModulesRead][key], _ = json.Marshal(catalog.ModulesByModuleValue{V: 1, None: "nothing is built of it", Manifest: digest, Repository: "novox/mesh-catalog", Commit: "c"})
var whole map[string]any
_ = json.Unmarshal([]byte(manifest), &whole)
f.entries[module] = catalog.CatalogEntry{Module: module, Manifest: whole}
}
return b
return f
}
func resolving(name string) []string {
@@ -97,7 +106,7 @@ func resolving(name string) []string {
// controller: the seats' verbs, the placed modules' own tools, every machine with its domains and addresses — a
// loopback address never one of them.
func TestTheMeshsToolsAreReadFromTheBus(t *testing.T) {
m, err := ReadMeshToolsFrom(aMesh(t), someSeats(), resolving)
m, err := ReadMeshToolsFrom(aMesh(t), aCatalog(), someSeats(), resolving)
if err != nil {
t.Fatal(err)
}
@@ -134,7 +143,7 @@ func TestTheMeshsToolsAreReadFromTheBus(t *testing.T) {
case <-time.After(10 * time.Second):
t.Fatalf("the seat list was not read: %+v", seatList.Status())
}
got, err := ReadMeshToolsFrom(aMesh(t), definedSeats(), resolving)
got, err := ReadMeshToolsFrom(aMesh(t), aCatalog(), definedSeats(), resolving)
var addrs []string
for _, r := range got.Replacements {
addrs = append(addrs, r.Address)
@@ -156,7 +165,7 @@ func TestTheMeshsToolsAreReadFromTheBus(t *testing.T) {
// **The table is generated, not written** (novox/hq ADR 0245): what the records say a verb replaces is a row,
// the machines' seats first, and nothing is rendered where the records say nothing.
func TestTheInsteadOfTableIsGeneratedFromTheRecords(t *testing.T) {
m, _ := ReadMeshToolsFrom(aMesh(t), someSeats(), resolving)
m, _ := ReadMeshToolsFrom(aMesh(t), aCatalog(), someSeats(), resolving)
table := InsteadOf(&m)
for _, want := range []string{
"| `<node>/node-service-manager.` | `status` — `systemctl status`, `systemctl is-active` · `journal` — `journalctl` |",
@@ -206,7 +215,7 @@ func TestTheInsteadOfTableIsGeneratedFromTheRecords(t *testing.T) {
// Asked again with the same answer, nothing is written and nothing renders.
func TestTheMeshsToolsAreKeptOnlyWhenTheyChange(t *testing.T) {
p := Paths{State: t.TempDir()}
m, _ := ReadMeshToolsFrom(aMesh(t), someSeats(), resolving)
m, _ := ReadMeshToolsFrom(aMesh(t), aCatalog(), someSeats(), resolving)
if !KeepMeshTools(p, m) {
t.Fatal("the first answer was not kept")
}
@@ -243,7 +252,7 @@ func TestTheMachinesRunningAModuleAreReadFromAssignments(t *testing.T) {
func TestTheMeshIsReadFromTheOwnersFirstViews(t *testing.T) {
bus := aMesh(t)
r := newMeshViews()
for _, name := range []string{ModulesRead, ManifestsRead, AssignmentsRead} {
for _, name := range []string{AssignmentsRead} {
state := strings.TrimPrefix(name, strings.Split(name, ".")[0]+".")
values := map[string]json.RawMessage{}
for k, v := range bus[name] {
@@ -257,16 +266,20 @@ func TestTheMeshIsReadFromTheOwnersFirstViews(t *testing.T) {
t.Fatal(err)
}
}
want, err := ReadMeshToolsFrom(bus, someSeats(), resolving)
want, err := ReadMeshToolsFrom(bus, aCatalog(), someSeats(), resolving)
if err != nil {
t.Fatal(err)
}
got, err := ReadMeshToolsFrom(viewStates{r}, someSeats(), resolving)
got, err := ReadMeshToolsFrom(viewStates{r}, aCatalog(), someSeats(), resolving)
if err != nil || !reflect.DeepEqual(got, want) || len(got.Replacements) == 0 {
t.Fatalf("from the views: %+v, %v\nfrom the state: %+v", got, err, want)
}
// A reader not yet answered says so, and reads nothing.
if _, err := ReadMeshToolsFrom(viewStates{newMeshViews()}, someSeats(), resolving); err == nil || !strings.Contains(err.Error(), "no first view") {
// A catalog half with no first view says so, and reads no module as absent.
if _, err := ReadMeshToolsFrom(bus, fakeCatalog{notYet: errors.New("the catalog half has no first view of main yet")}, someSeats(), resolving); err == nil || !strings.Contains(err.Error(), "no first view") {
t.Fatalf("an unanswered catalog half: %v", err)
}
if _, err := ReadMeshToolsFrom(viewStates{newMeshViews()}, aCatalog(), someSeats(), resolving); err == nil || !strings.Contains(err.Error(), "no first view") {
t.Fatalf("an unanswered first view: %v", err)
}
}
@@ -274,7 +287,7 @@ func TestTheMeshIsReadFromTheOwnersFirstViews(t *testing.T) {
// Issue 1126 B13 AC1 (ADR 0590): over the seats the mesh defines, the rendered instructions, their table among them,
// name no address of the controller, which is not placed: each job is named by the seat that owns it now.
func TestTheRenderedInstructionsNameNoControllerAddress(t *testing.T) {
m, err := ReadMeshToolsFrom(aMesh(t), someSeats(), resolving)
m, err := ReadMeshToolsFrom(aMesh(t), aCatalog(), someSeats(), resolving)
if err != nil {
t.Fatal(err)
}
modules/claude-code/cmd/claude-code/main.go
+26 -6
View File
@@ -25,6 +25,7 @@ import (
condseat "git.novox.be/novox/mesh-contracts/health-and-repair/go/seats/meshconditions"
occ "git.novox.be/novox/mesh-contracts/operator-and-conversation/go/clients/operatorchannelclient"
oc "git.novox.be/novox/mesh-contracts/operator-and-conversation/go/seats/operatorchannel"
catalogclient "git.novox.be/novox/mesh-delivery/contracts/go/clients/meshcatalogclient"
stdio "git.novox.be/novox/mesh-sdk/go"
"git.novox.be/novox/mesh-sdk/go/seatlist"
"git.novox.be/novox/mesh-sdk/go/views"
@@ -217,12 +218,19 @@ func nodesRunningMe() ([]string, error) { return NodesRunning(viewStates{meshVie
// holds them equal).
const (
assignerModule = "mesh-delivery-assigner"
indexerModule = "mesh-delivery-indexer"
)
// The seat verbs the follower of the watched branches asks (design 47 G §614), as the manifest's `invokes` names them.
const (
RevisionVerb = "seat:mesh-catalog.revision"
RegistrationsVerb = "seat:mesh-catalog.registrations"
SourceVerb = "seat:mesh-delivery.source"
)
// meshCatalog is this module's reading of which modules exist: the catalog half, set where the reader is built.
var meshCatalog CatalogHalf
const (
ModulesTool = indexerModule + ".indexer_read_modules"
ManifestsTool = indexerModule + ".indexer_read_manifests"
AssignmentsTool = assignerModule + ".assigner_read_assignments"
)
@@ -253,8 +261,6 @@ func newMeshViews() *views.Reader {
default:
}
}}
v.Follow(indexerModule, "modules", ModulesTool)
v.Follow(indexerModule, "manifests", ManifestsTool)
v.Follow(assignerModule, "assignments", AssignmentsTool)
return v
}
@@ -335,7 +341,7 @@ func tools(p Paths, servers ServerState, view *ServerView, config ConfigState, c
Description: "Write Claude Code's managed directory now, from the mesh's facts, this module's settings and the servers registered here — after reading the bus again for what each tool replaces and which machines the mesh has (the instructions' table and the shell's guard).",
Run: func(map[string]any) (any, error) {
answer := map[string]any{}
if m, err := ReadMeshToolsFrom(viewStates{meshViews}, definedSeats(), lookup); err != nil {
if m, err := ReadMeshToolsFrom(viewStates{meshViews}, meshCatalog, definedSeats(), lookup); err != nil {
answer["asked"] = "the bus could not be read in full (" + err.Error() + "); what was kept is rendered"
} else {
KeepMeshTools(p, m)
@@ -469,6 +475,20 @@ func main() {
}
p, launched := PathsFrom(os.Getenv)
meshViews = newMeshViews()
branches, err := catalogclient.FollowBranches(meshViews, func(key string, body any) (json.RawMessage, error) { return stdio.Ask(key, body) })
if err != nil {
say("%v", err)
os.Exit(1)
}
branches.Logf = say
// A change in a branch's half is told as any followed state's: the instructions' table is read again.
branches.Changed = func(string, string, json.RawMessage, bool, time.Time) {
select {
case meshChanged <- struct{}{}:
default:
}
}
meshCatalog = branchesHalf{branches}
if launched {
fsys = FilesFor(p)
}
modules/claude-code/cmd/claude-code/mesh.go
+13 -33
View File
@@ -18,8 +18,6 @@ import (
"fmt"
assignerclient "git.novox.be/novox/mesh-contracts/go/clients/meshmoduleassignerclient"
mesh "git.novox.be/novox/mesh-contracts/go/seats/meshmoduleassigner"
catalogclient "git.novox.be/novox/mesh-delivery/contracts/go/clients/meshcatalogclient"
catalog "git.novox.be/novox/mesh-delivery/contracts/go/seats/meshcatalog"
"net"
"os"
"path/filepath"
@@ -93,8 +91,6 @@ func ReadMeshTools(p Paths) *MeshTools {
// The states read, each declared in the manifest's reads.
const (
AssignmentsRead = (assignerModule + "." + mesh.StateAssignments)
ModulesRead = indexerModule + "." + catalog.StateModules
ManifestsRead = indexerModule + "." + catalog.StateManifests
)
// States reads the module states the manifest declares in its reads: a state's keys, and one key's value.
@@ -152,44 +148,28 @@ func NodesRunning(st States, module string) ([]string, error) {
return runsOn(placed)[module], nil
}
// modulesTools is every placed module's own tools that replace a command, by module, from the manifest the indexer
// keeps of it. A module whose manifest cannot be read is left out and said in the error.
func modulesTools(st States, on map[string][]string) ([]ModuleTools, error) {
// modulesTools is every placed module's own tools that replace a command, by module, from the manifest its catalog
// graph entry carries. A module whose manifest cannot be read is left out and said in the error.
func modulesTools(cat CatalogHalf, on map[string][]string) ([]ModuleTools, error) {
var out []ModuleTools
var problems []error
if err := cat.Ready(); err != nil {
return nil, err
}
modules := make([]string, 0, len(on))
for m := range on {
modules = append(modules, m)
}
sort.Strings(modules)
for _, module := range modules {
key, err := catalogclient.ModulesKey(module)
if err != nil {
continue
}
raw, found, err := st.Get(ModulesRead, key)
entry, found, err := cat.Entry(module)
if err != nil || !found {
if err != nil {
problems = append(problems, fmt.Errorf("%s %s: %w", ModulesRead, module, err))
problems = append(problems, fmt.Errorf("the catalog half's entry of %s: %w", module, err))
}
continue
}
m, err := catalogclient.DecodeModule(raw)
if err != nil || m.Manifest == "" {
continue
}
mkey, err := catalogclient.ManifestsKey(m.Manifest)
if err != nil {
continue
}
raw, found, err = st.Get(ManifestsRead, mkey)
if err != nil || !found {
if err != nil {
problems = append(problems, fmt.Errorf("%s of %s: %w", ManifestsRead, module, err))
}
continue
}
kept, err := catalogclient.DecodeManifest(raw)
kept, err := json.Marshal(entry.Manifest)
if err != nil {
problems = append(problems, fmt.Errorf("the manifest of %s: %w", module, err))
continue
@@ -197,7 +177,7 @@ func modulesTools(st States, on map[string][]string) ([]ModuleTools, error) {
var man struct {
Replaces map[string][]string `json:"replaces"`
}
if json.Unmarshal([]byte(kept.Manifest), &man) != nil || len(man.Replaces) == 0 {
if json.Unmarshal(kept, &man) != nil || len(man.Replaces) == 0 {
continue
}
out = append(out, ModuleTools{Module: module, On: on[module], Replaces: man.Replaces})
@@ -263,14 +243,14 @@ func lookup(name string) []string {
// ReadMeshToolsFrom reads the bus now: the seats' verbs, the placed modules' own tools, and every node assignments names
// with its domains and addresses. A part that cannot be read is left out and said in the error; what was read is kept.
func ReadMeshToolsFrom(st States, defined []seatdef.Definition, resolve func(string) []string) (MeshTools, error) {
func ReadMeshToolsFrom(st States, cat CatalogHalf, defined []seatdef.Definition, resolve func(string) []string) (MeshTools, error) {
var m MeshTools
placed, err := assignments(st)
if err != nil {
m.Replacements = ReplacementsOf(defined, nil)
return m, err
}
modules, merr := modulesTools(st, runsOn(placed))
modules, merr := modulesTools(cat, runsOn(placed))
m.Replacements = ReplacementsOf(defined, modules)
for node, p := range placed {
machine := Machine{Name: node, Domains: []string{node + InternalSuffix}}
@@ -357,7 +337,7 @@ func refreshMeshTools(p Paths) {
}
startSeatList(seatlist.StdioBus())
for {
m, err := ReadMeshToolsFrom(viewStates{meshViews}, definedSeats(), lookup)
m, err := ReadMeshToolsFrom(viewStates{meshViews}, meshCatalog, definedSeats(), lookup)
if err != nil {
say("reading what each tool replaces and which machines the mesh has: %v", err)
}
modules/claude-code/module.json
+6 -4
View File
@@ -45,13 +45,15 @@
"seat:mesh-hand-acts.warranted",
"seat:mesh-hand-acts.hand-act",
"seat:mesh-catalog.seat-definitions",
"mesh-delivery-indexer.indexer_read_modules",
"mesh-delivery-indexer.indexer_read_manifests",
"seat:mesh-catalog.revision",
"seat:mesh-catalog.registrations",
"seat:mesh-delivery.source",
"mesh-delivery-assigner.assigner_read_assignments"
],
"consumes": [
"mesh-delivery-indexer.views.modules",
"mesh-delivery-indexer.views.manifests",
"mesh-catalog.catalog-revision",
"mesh-catalog.registration-changed",
"mesh-delivery.source-moved",
"mesh-delivery-assigner.views.assignments",
"mesh-catalog.seat-definition-changed",
"mesh-catalog.seat-definition-removed",
modules/mesh-build-go/cmd/mesh-build-go/testdata/repos/probe/seen.json
Vendored
+1
View File
@@ -0,0 +1 @@
{"key":false,"login":false,"resolved":true,"socket":false,"tcp":false,"wroteCache":false}
modules/mesh-config/cmd/mesh-config/catalogfeed.go
+68
View File
@@ -0,0 +1,68 @@
package main
// mesh-config's reading of each module's declarations (design 47 G §196, §594, §614): the catalog
// half, taken through meshcatalogclient's follower of the watched branches on the part's one views reader. A module's
// manifest is the one its catalog graph entry carries whole (§443), never the indexer's state `modules` or `manifests`; the
// manifest a running build was made from is the entry read by that build's fingerprint (§614, an older build's entry).
import (
"context"
"fmt"
"strings"
"git.novox.be/novox/mesh-delivery/contracts/go/clients/meshcatalogclient"
catalog "git.novox.be/novox/mesh-delivery/contracts/go/seats/meshcatalog"
)
// CatalogHalf is what the holder reads of the catalog half.
type CatalogHalf interface {
// Entry is a module's entry at the trunk; an error while a watched branch has no first view yet, never "no module".
Entry(module string) (catalog.CatalogEntry, bool, error)
// Ready is nil once every watched branch has its first view (none-yet counting as one), else says what it waits for.
Ready() error
// ByFingerprint reads an entry by the fingerprint of a build: from a branch's half when one lists it, else asked of the
// seat's verb `entry` once and kept (asked, answered or not found).
ByFingerprint(ctx context.Context, fingerprint string) (catalog.CatalogEntry, meshcatalogclient.EntryState)
}
// branchesHalf is the CatalogHalf over the follower of the watched branches.
type branchesHalf struct{ b *meshcatalogclient.Branches }
func (h branchesHalf) Ready() error {
followed := h.b.Followed()
if len(followed) == 0 {
return fmt.Errorf("the catalog half names no watched branch yet (the seat mesh-catalog's registrations are asked until they answer)")
}
var waiting []string
for _, branch := range followed {
if c, ok := h.b.Catalog(branch); ok {
select {
case <-c.Ready():
default:
waiting = append(waiting, branch)
}
}
}
if len(waiting) > 0 {
return fmt.Errorf("the catalog half has no first view of %s yet", strings.Join(waiting, ", "))
}
return nil
}
func (h branchesHalf) Entry(module string) (catalog.CatalogEntry, bool, error) {
if err := h.Ready(); err != nil {
return catalog.CatalogEntry{}, false, err
}
for _, branch := range h.b.Followed() { // the branch followed first names a module more than one hold
if c, ok := h.b.Catalog(branch); ok {
if e, ok := c.Entry(module); ok {
return e, true, nil
}
}
}
return catalog.CatalogEntry{}, false, nil
}
func (h branchesHalf) ByFingerprint(ctx context.Context, fingerprint string) (catalog.CatalogEntry, meshcatalogclient.EntryState) {
return h.b.Resolve(ctx, fingerprint)
}
modules/mesh-config/cmd/mesh-config/catalogfeed_test.go
+90
View File
@@ -0,0 +1,90 @@
package main
import (
"context"
"encoding/json"
"fmt"
"strings"
"sync"
"testing"
"time"
"git.novox.be/novox/mesh-delivery/contracts/go/clients/meshcatalogclient"
catalog "git.novox.be/novox/mesh-delivery/contracts/go/seats/meshcatalog"
delivery "git.novox.be/novox/mesh-delivery/contracts/go/seats/meshdelivery"
"git.novox.be/novox/mesh-sdk/go/views"
)
// wiringOf is a part's reader in miniature: it keeps the followers adopted, so a test starts them as the reader's Start does.
type wiringOf struct {
mu sync.Mutex
fs []*views.Follower
}
func (w *wiringOf) Adopt(f *views.Follower, _, _ string, _ func(context.Context, []byte) error) error {
w.mu.Lock()
defer w.mu.Unlock()
w.fs = append(w.fs, f)
return nil
}
func (w *wiringOf) Drop(string, string) error { return nil }
// Design 47 G §196, §594: which modules exist, and the manifest of each, are the catalog half's; a branch with no first
// view yet is an error naming it, never no module, and an entry's manifest is the one it carries whole.
func TestCatalogHalfIsTheWatchedBranchesNewestRevision(t *testing.T) {
ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
defer cancel()
answered := false
var mu sync.Mutex
call := func(key string, _ any) (json.RawMessage, error) {
mu.Lock()
defer mu.Unlock()
switch key {
case meshcatalogclient.RegistrationsVerb:
return json.Marshal(catalog.RegistrationsAnswer{Registrations: []catalog.Registration{{Module: "m", Version: 1, Branches: []string{"main"}}}})
case meshcatalogclient.SourceVerb:
return json.Marshal(delivery.SourceAnswer{})
case meshcatalogclient.RevisionVerb:
if !answered {
return nil, fmt.Errorf("the holder is away")
}
return json.Marshal(catalog.RevisionAnswer{Revision: &catalog.CatalogRevision{Revision: 1, Kind: meshcatalogclient.KindTrunk},
Entries: []catalog.CatalogEntry{{Module: "pg", Manifest: map[string]any{"module": "pg"}}}})
}
return nil, fmt.Errorf("asked %s", key)
}
w := &wiringOf{}
b, err := meshcatalogclient.NewBranches(w, call)
if err != nil {
t.Fatal(err)
}
h := branchesHalf{b}
if _, _, err := h.Entry("pg"); err == nil || !strings.Contains(err.Error(), "no watched branch") {
t.Fatalf("before the registrations: %v, want an error naming the missing branch", err)
}
go func() { _ = w.fs[0].Start(ctx) }() // the registrations
for len(b.Followed()) == 0 {
select {
case <-ctx.Done():
t.Fatal("the registrations' branch is never followed")
case <-time.After(10 * time.Millisecond):
}
}
if _, _, err := h.Entry("pg"); err == nil || !strings.Contains(err.Error(), "no first view of main") {
t.Fatalf("with no first view: %v, want an error naming the branch, never no module", err)
}
mu.Lock()
answered = true
mu.Unlock()
c, _ := b.Catalog("main")
if err := c.Start(ctx); err != nil {
t.Fatal(err)
}
got, ok, err := h.Entry("pg")
if err != nil || !ok || got.Manifest["module"] != "pg" || h.Ready() != nil {
t.Fatalf("with the first view: %v, %v, %v", got, ok, err)
}
if _, ok, err := h.Entry("absent"); ok || err != nil {
t.Fatalf("a module the half does not list: %v, %v", ok, err)
}
}
modules/mesh-config/cmd/mesh-config/config.go
+3
View File
@@ -98,6 +98,9 @@ type Config struct {
Forge PullForge
// Delivery is what the delivery parts answer (renames.go, retiring.go); nil, no window closes and nothing retires.
Delivery Delivery
// Catalog is the catalog half: which modules exist and the manifest each entry carries, and the entry of a build's
// fingerprint (design 47 G §196, §594).
Catalog CatalogHalf
// Vault is the seat mesh-vault, for the retiring of a secret; nil, a secret does not retire.
Vault Vault
// RouterWithin and RouterEvery bound how long a proposal waits for the router's word on its ask, and how often
modules/mesh-config/cmd/mesh-config/config_test.go
+55 -27
View File
@@ -1,6 +1,7 @@
package main
import (
"context"
"encoding/json"
"errors"
"fmt"
@@ -8,7 +9,6 @@ import (
meshseat "git.novox.be/novox/mesh-contracts/go/seats/meshmoduleassigner"
catalogclient "git.novox.be/novox/mesh-delivery/contracts/go/clients/meshcatalogclient"
catalog "git.novox.be/novox/mesh-delivery/contracts/go/seats/meshcatalog"
builds "git.novox.be/novox/mesh-delivery/modules/mesh-delivery-builder/builds"
"sort"
"strings"
"sync"
@@ -27,10 +27,34 @@ type memStore struct {
handActs []string
// failHandAct makes the hand-act log refuse.
failHandAct bool
// entries are the catalog half's trunk entries by module, and byFp the entries a build's fingerprint names: the memStore
// is also the CatalogHalf the tests give (catalogfeed.go).
entries map[string]catalog.CatalogEntry
byFp map[string]catalog.CatalogEntry
notYet error
}
func (m *memStore) Entry(module string) (catalog.CatalogEntry, bool, error) {
m.mu.Lock()
defer m.mu.Unlock()
e, ok := m.entries[module]
return e, ok, m.notYet
}
func (m *memStore) Ready() error { return m.notYet }
func (m *memStore) ByFingerprint(_ context.Context, fp string) (catalog.CatalogEntry, catalogclient.EntryState) {
m.mu.Lock()
defer m.mu.Unlock()
if e, ok := m.byFp[fp]; ok {
return e, catalogclient.EntryAnswered
}
return catalog.CatalogEntry{}, catalogclient.EntryNotFound
}
func newStore() *memStore {
return &memStore{states: map[string]map[string][]byte{}, logs: map[string][]LogEntry{}}
return &memStore{states: map[string]map[string][]byte{}, logs: map[string][]LogEntry{}, entries: map[string]catalog.CatalogEntry{},
byFp: map[string]catalog.CatalogEntry{}}
}
func (m *memStore) Get(state, key string) (*Entry, error) {
@@ -206,38 +230,31 @@ func (m *memStore) keysOf(state string) []string {
return keys
}
// manifest keeps a manifest in the indexer's manifests by its digest, as the indexer writes it, and answers the digest.
func (m *memStore) manifest(t *testing.T, module, raw string) string {
// entryOf is a module's catalog entry carrying the manifest raw whole, with its digest and commit.
func entryOf(t *testing.T, module, raw, fp string) catalog.CatalogEntry {
t.Helper()
if !json.Valid([]byte(raw)) {
var whole map[string]any
if err := json.Unmarshal([]byte(raw), &whole); err != nil {
t.Fatalf("not JSON: %s", raw)
}
digest := catalogclient.ManifestDigest([]byte(raw))
value, err := catalogclient.EncodeManifest(catalog.ManifestsByDigestValue{Module: module, Manifest: raw})
if err != nil {
t.Fatal(err)
}
m.mirror(t, ManifestsRead, digest, string(value))
return digest
return catalog.CatalogEntry{Module: module, Fp: fp, Manifest: whole, ManifestDigest: catalogclient.ManifestDigest([]byte(raw)), Commit: "c1"}
}
// declare makes raw the module's manifest at the trunk: kept by digest, and named by the indexer's modules.<module>.
// declare makes raw the module's manifest at the trunk: the entry the catalog half holds of it.
func (m *memStore) declare(t *testing.T, module, raw string) {
t.Helper()
digest := m.manifest(t, module, raw)
m.mirror(t, ModulesRead, module, fmt.Sprintf(`{"v":1,"manifest":%q,"repository":"novox/%s","commit":"c1"}`,
digest, module))
m.mu.Lock()
defer m.mu.Unlock()
m.entries[module] = entryOf(t, module, raw, "")
}
// want makes a node's want name a build of the module whose manifest is raw, as the assigner and the builder keep them.
// want makes a node's want name a build of the module whose manifest is raw, as the assigner keeps it and the catalog half
// holds the entry of that build's fingerprint.
func (m *memStore) want(t *testing.T, node, module, fp, raw string) {
t.Helper()
digest := m.manifest(t, module, raw)
key, err := builds.BuilderBuildKey(module, fp)
if err != nil {
t.Fatal(err)
}
m.mirror(t, BuildsRead, key, fmt.Sprintf(`{"v":1,"state":"done","manifest":%q}`, digest))
m.mu.Lock()
m.byFp[fp] = entryOf(t, module, raw, fp)
m.mu.Unlock()
m.mirror(t, WantsRead, "want."+node+"."+module, fmt.Sprintf(`{"v":1,"fp":%q,"self":true}`, fp))
}
@@ -280,7 +297,7 @@ func aMesh(t *testing.T) (*Config, *memStore) {
m.assign(t, "g14", "claude-code")
m.assign(t, "g14", "old")
clock := time.Date(2026, 10, 14, 19, 0, 0, 0, time.UTC)
c := &Config{Store: m, Now: func() time.Time { clock = clock.Add(time.Second); return clock }, Logf: t.Logf}
c := &Config{Store: m, Catalog: m, Now: func() time.Time { clock = clock.Add(time.Second); return clock }, Logf: t.Logf}
if _, err := c.Reconcile(); err != nil {
t.Fatal(err)
}
@@ -712,21 +729,32 @@ func TestDeclarationsComeFromTheManifestsAsBuilt(t *testing.T) {
m.mirror(t, WantsRead, "want.novox.de-spiegel", `{"v":1,"removed":true}`)
},
resolved: map[string]string{"de-spiegel.praktijk-name": "unset "}, absent: []string{"de-spiegel.old-key"}},
{name: "a placed module the indexer holds no manifest of is unresolved, said, its values kept",
{name: "a placed module the catalog half holds no entry of is unresolved, said, its values kept",
given: func(t *testing.T, m *memStore) {
must(t, m.Put(ResolvedState, ResolvedKey("novox", "de-spiegel", "praktijk-name"),
Resolved{Node: "novox", Module: "de-spiegel", Key: "praktijk-name", Source: SourceUnset}))
m.assign(t, "novox", "de-spiegel")
},
resolved: map[string]string{"de-spiegel.praktijk-name": "unset "},
unresolved: "the indexer holds no modules.de-spiegel"},
unresolved: "the catalog half holds no entry of de-spiegel"},
// With no first view of the catalog half a module is never read as absent: the error names the view (design 47 G §196).
{name: "a catalog half with no first view leaves its module unresolved, said, its values kept",
given: func(t *testing.T, m *memStore) {
m.declare(t, "de-spiegel", despiegel)
m.notYet = errors.New("the catalog half has no first view of main yet")
must(t, m.Put(ResolvedState, ResolvedKey("novox", "de-spiegel", "praktijk-name"),
Resolved{Node: "novox", Module: "de-spiegel", Key: "praktijk-name", Source: SourceUnset}))
m.assign(t, "novox", "de-spiegel")
},
resolved: map[string]string{"de-spiegel.praktijk-name": "unset "},
unresolved: "no first view of main"},
}
for _, tc := range cases {
t.Run(tc.name, func(t *testing.T) {
m := newStore()
tc.given(t, m)
var said []string
c := &Config{Store: m, Now: time.Now, Logf: func(f string, a ...any) { said = append(said, fmt.Sprintf(f, a...)) }}
c := &Config{Store: m, Catalog: m, Now: time.Now, Logf: func(f string, a ...any) { said = append(said, fmt.Sprintf(f, a...)) }}
if _, err := c.Reconcile(); err != nil {
t.Fatal(err)
}
modules/mesh-config/cmd/mesh-config/main.go
+31
View File
@@ -21,6 +21,7 @@ import (
"git.novox.be/novox/mesh-contracts/health-and-repair/go/clients/meshhandactsclient"
meshconditions "git.novox.be/novox/mesh-contracts/health-and-repair/go/seats/meshconditions"
oc "git.novox.be/novox/mesh-contracts/operator-and-conversation/go/seats/operatorchannel"
catalogclient "git.novox.be/novox/mesh-delivery/contracts/go/clients/meshcatalogclient"
stdio "git.novox.be/novox/mesh-sdk/go"
"git.novox.be/novox/mesh-sdk/go/views"
)
@@ -137,6 +138,36 @@ func main() {
default:
}
}, w)
// Which modules exist, each one's manifest and a build's are the catalog half's, followed per watched branch by the
// seat's client on the same reader (design 47 G §196, §614).
branches, err := catalogclient.FollowBranches(theViews.reader, stdio.Ask)
if err != nil {
logf("%v", err)
os.Exit(1)
}
branches.Logf = logf
branches.Changed = func(string, string, json.RawMessage, bool, time.Time) {
select {
case moved <- struct{}{}:
default:
}
}
branches.Entries.Changed = func(catalogclient.EntryRow) {
select {
case moved <- struct{}{}:
default:
}
}
c.Catalog = branchesHalf{branches}
theViews.catalog = func() []*views.Follower {
out := []*views.Follower{branches.Registrations.Follower}
for _, b := range branches.Followed() {
if cat, ok := branches.Catalog(b); ok {
out = append(out, cat.Follower)
}
}
return out
}
// the runtime answers the writer's subscription only once Serve has its channel
go func() {
<-stdio.Served()
modules/mesh-config/cmd/mesh-config/manifest_test.go
+19 -9
View File
@@ -23,6 +23,8 @@ func TestEveryStateAndActHasItsTool(t *testing.T) {
Packages map[string]string `json:"packages"`
Logs []string `json:"logs"`
Reads []string `json:"reads"`
Invokes []string `json:"invokes"`
Consumes []string `json:"consumes"`
Tools []string `json:"tools"`
Claims []struct {
Name string `json:"name"`
@@ -32,16 +34,24 @@ func TestEveryStateAndActHasItsTool(t *testing.T) {
if err := json.Unmarshal(raw, &m); err != nil {
t.Fatal(err)
}
// The builder's keys are read through its published package, taken by its series (ADR 0634 §1).
if m.Packages["git.novox.be/novox/mesh-delivery/modules/mesh-delivery-builder/builds"] != "0.1" {
t.Errorf("packages %v: want builds of series 0.1", m.Packages)
// Which modules exist and a build's manifest are the catalog half's (design 47 G §594): no builder package, and the
// manifest names neither the indexer's modules and manifests nor the builder's builds.
if len(m.Packages) != 0 {
t.Errorf("packages %v: the builder's keys are no longer read", m.Packages)
}
for _, list := range [][]string{m.Reads, m.Invokes, m.Consumes} {
for _, e := range list {
if strings.Contains(e, "mesh-delivery-builder") || strings.HasSuffix(e, "indexer_read_modules") || strings.HasSuffix(e, "indexer_read_manifests") ||
strings.HasSuffix(e, "views.modules") || strings.HasSuffix(e, "views.manifests") {
t.Errorf("the manifest names %s, which only the catalog half answers now", e)
}
}
}
via := map[string]string{
ValuesState: "config list", ResolvedState: "config list", PredefinedState: "config list",
ProposalsState: "config proposals", PullsState: "config list", KeptState: "config list", HistoryLog: "config history",
DeclaredState: DeclaredTool,
AssignmentsRead: "config list", WantsRead: "config list", BuildsRead: "config list", ModulesRead: DeclaredTool,
ManifestsRead: DeclaredTool, IndexerPredefinedRead: "config list", CandidatesRead: ResyncTool,
AssignmentsRead: "config list", WantsRead: "config list", IndexerPredefinedRead: "config list", CandidatesRead: ResyncTool,
"a resolution": ResyncTool, "the indexer's candidates taken": ResyncTool,
// design 32: each new state with the tool that reads it, each act with the tool that starts it.
RenamesState: RenamesTool, RetiringState: RetiringTool, FindingsState: "config_renames and config_retiring",
@@ -80,10 +90,10 @@ func TestEveryStateAndActHasItsTool(t *testing.T) {
t.Errorf("the manifest lists the tool %s, which is not served", name)
}
}
// The delivery owners' state is read through their tools and views (ADR 0615 §1): only the builder's builds, which has
// no tool yet, is still a read.
if len(m.Reads) != 1 || m.Reads[0] != BuildsRead {
t.Errorf("reads %v, want the builder's builds alone", m.Reads)
// The delivery owners' state is read through their tools and views (ADR 0615 §1), and which modules exist through the
// catalog half (design 47 G §196): the manifest reads no state of theirs.
if len(m.Reads) != 0 {
t.Errorf("reads %v, want none", m.Reads)
}
for _, list := range [][]string{m.State, m.Logs, m.Reads} {
for _, s := range list {
modules/mesh-config/cmd/mesh-config/ownerviews.go
+32 -14
View File
@@ -1,14 +1,15 @@
package main
// What mesh-config reads of the delivery owners' state, through their tools and views families (novox/hq ADR 0615 §1,
// §4 to §6, ADR 0596): never a bucket of theirs. For each of the six states it follows (the indexer's modules, manifests,
// predefined and config-candidates, the assigner's assignments and wants) it asks the owner's read tool for the whole
// §4 to §6, ADR 0596): never a bucket of theirs. For each of the four states it follows (the indexer's predefined and
// config-candidates, the assigner's assignments and wants) it asks the owner's read tool for the whole
// state at its start, keeps the picture from the owner's `views.<state>` events, and asks again only on a gap. Started
// while an owner is down it keeps asking, and past two minutes writes a finding naming the owner and the tool.
//
// The ordinary read tools are asked, not the first-view tools: those answer a few named callers (mesh-grants,
// node-tools, the delivery parts), and mesh-config is not one. The builder's `builds` has no owner tool yet (A2 built the
// indexer's and the assigner's), so it is still read as a `reads` entry.
// node-tools, the delivery parts), and mesh-config is not one. Which modules exist, and each one's manifest, are not the
// indexer's: they are the catalog half's, followed per watched branch by meshcatalogclient on the same reader (design 47 G
// §196, §594, §614; catalogfeed.go).
import (
"context"
@@ -25,8 +26,6 @@ import (
// The tools of the owners that mesh-config asks, by the state they read.
const (
IndexerModulesTool = "indexer_read_modules"
IndexerManifestsTool = "indexer_read_manifests"
IndexerPredefTool = "indexer_read_predefined"
IndexerCandidatesTool = "indexer_read_config_candidates"
AssignerAssignmentsTool = "assigner_read_assignments"
@@ -39,8 +38,6 @@ const (
// followed is each state read through its owner: the `<owner>.<state>` name this module's code uses, the owner, its
// state and its tool.
var followed = []struct{ Read, Owner, State, Tool string }{
{ModulesRead, indexerModule, "modules", IndexerModulesTool},
{ManifestsRead, indexerModule, "manifests", IndexerManifestsTool},
{IndexerPredefinedRead, indexerModule, catalog.StatePredefined, IndexerPredefTool},
{CandidatesRead, indexerModule, catalog.StateConfigCandidates, IndexerCandidatesTool},
{AssignmentsRead, assignerModule, "assignments", AssignerAssignmentsTool},
@@ -53,6 +50,9 @@ var followed = []struct{ Read, Owner, State, Tool string }{
type ownerViews struct {
by map[string]*views.Follower
reader *views.Reader
// catalog is the followers of the catalog half (the registrations and each watched branch's revision view), which the
// first-view tools list and ask again beside the owners' (design 47 G §614); nil before they are followed.
catalog func() []*views.Follower
}
// newOwnerViews makes a follower for each state; changed is told when any picture changes. findings is the module's writer,
@@ -137,26 +137,43 @@ func (o *ownerViews) keys(state string) ([]string, bool, error) {
return out, true, nil
}
// statuses are the first views' statuses, by owner and state.
// catalogFollowers are the catalog half's followers, none before they are followed.
func (o *ownerViews) catalogFollowers() []*views.Follower {
if o.catalog == nil {
return nil
}
return o.catalog()
}
// statuses are the first views' statuses, by owner and state: the owners', then the catalog half's.
func (o *ownerViews) statuses() []views.Status {
var out []views.Status
for _, f := range followed {
out = append(out, o.by[f.Read].Status())
}
for _, f := range o.catalogFollowers() {
out = append(out, f.Status())
}
return out
}
// again asks one first view once more by hand ("<owner>.<state>", or all when empty).
func (o *ownerViews) again(ctx context.Context, which, why string) ([]string, error) {
var done []string
all := make([]*views.Follower, 0, len(followed))
for _, f := range followed {
if which != "" && which != f.Owner+"."+f.State {
all = append(all, o.by[f.Read])
}
all = append(all, o.catalogFollowers()...)
for _, f := range all {
name := f.Owner + "." + f.State
if which != "" && which != name {
continue
}
if err := o.by[f.Read].Again(ctx, why); err != nil {
if err := f.Again(ctx, why); err != nil {
return done, err
}
done = append(done, f.Owner+"."+f.State)
done = append(done, name)
}
if len(done) == 0 {
return nil, errors.New("no first view is named " + which)
@@ -168,8 +185,9 @@ func (o *ownerViews) again(ctx context.Context, which, why string) ([]string, er
func viewTools(c *Config, o func() *ownerViews) []stdio.Tool {
return []stdio.Tool{
{Name: FirstViewsTool,
Description: "mesh-config's first views of the delivery owners' state: for each of the indexer's modules, manifests, " +
"predefined and config-candidates and the assigner's assignments and wants, the tool it asks, the revision its " +
Description: "mesh-config's first views of the delivery owners' state: for each of the indexer's " +
"predefined and config-candidates, the assigner's assignments and wants and the catalog half's registrations and " +
"each watched branch's revision, the tool it asks, the revision its " +
"picture holds, when it asked and was answered, how many asks (and by hand), and whether its condition is raised.",
Run: func(map[string]any) (any, error) {
ov := o()
modules/mesh-config/cmd/mesh-config/ownerviews_test.go
+6 -2
View File
@@ -39,8 +39,12 @@ func TestAFollowedStateIsReadFromItsOwnersFirstViewAndEvents(t *testing.T) {
if !o.reader.Follows("mesh-delivery-assigner.wants") {
t.Error("the wants view is not one of the reader's (A-28)")
}
if _, followed, _ := o.get(BuildsRead, "x"); followed {
t.Error("the builder's builds is followed, and it has no owner tool")
// The indexer's modules and manifests and the builder's builds are no states this module follows: which modules exist is
// the catalog half's (design 47 G §196).
for _, gone := range []string{"mesh-delivery-indexer.modules", "mesh-delivery-indexer.manifests", "mesh-delivery-builder.builds"} {
if _, followed, _ := o.get(gone, "x"); followed {
t.Errorf("%s is followed", gone)
}
}
ctx := context.Background()
go func() { _ = o.by[WantsRead].Start(ctx) }()
modules/mesh-config/cmd/mesh-config/renames_test.go
+2 -2
View File
@@ -80,7 +80,7 @@ func aRenamingMesh(t *testing.T) (*Config, *memStore, *fakeDelivery, *time.Time)
}
clock := time.Date(2026, 11, 4, 10, 0, 0, 0, time.UTC)
d := &fakeDelivery{haves: map[string]string{"g14.player": fp1, "novox.player": fp1}, notSays: map[string]bool{}}
c := &Config{Store: m, Now: func() time.Time { return clock }, Logf: t.Logf, Delivery: d, Findings: m}
c := &Config{Store: m, Catalog: m, Now: func() time.Time { return clock }, Logf: t.Logf, Delivery: d, Findings: m}
if _, err := c.Reconcile(); err != nil {
t.Fatal(err)
}
@@ -275,7 +275,7 @@ func TestAWindowHeldSevenDaysIsAFindingAndAHandCloseIsAHandAct(t *testing.T) {
}
m.handActs = nil
// A restart in between: the module holds no finding of its own, the writer holds it (design 47 G A-32, A-38).
c = &Config{Store: c.Store, Now: c.Now, Logf: c.Logf, Delivery: c.Delivery, Findings: c.Findings}
c = &Config{Store: c.Store, Catalog: c.Catalog, Now: c.Now, Logf: c.Logf, Delivery: c.Delivery, Findings: c.Findings}
if _, err := c.RenameClose("player", "level", "novox", "novox is retired for good"); err != nil {
t.Fatal(err)
}
modules/mesh-config/cmd/mesh-config/resolve.go
-8
View File
@@ -4,8 +4,6 @@ import (
"encoding/json"
"fmt"
meshseat "git.novox.be/novox/mesh-contracts/go/seats/meshmoduleassigner"
catalog "git.novox.be/novox/mesh-delivery/contracts/go/seats/meshcatalog"
builds "git.novox.be/novox/mesh-delivery/modules/mesh-delivery-builder/builds"
"sort"
"strings"
"time"
@@ -29,12 +27,6 @@ const (
AssignmentsRead = (assignerModule + "." + meshseat.StateAssignments)
// WantsRead is the assigner's wants, `want.<node>.<module>`: the build each node is to run (ADR 0539 §1).
WantsRead = assignerModule + ".wants"
// BuildsRead is the builder's builds, `<module>.<fingerprint hex>`: each build's manifest digest (ADR 0525).
BuildsRead = builds.BuilderModule + ".builds"
// ModulesRead is the indexer's modules, `<module>`: the trunk's manifest digest of each module, from whichever
// repository holds it (ADR 0437 §5, ADR 0478 §9); ManifestsRead its manifests by digest (ADR 0429 §16).
ModulesRead = indexerModule + "." + catalog.StateModules
ManifestsRead = indexerModule + "." + catalog.StateManifests
// HistoryLog is the log of every set and clear, with the proven caller and why (ADR 0348 §9).
HistoryLog = "history"
modules/mesh-config/cmd/mesh-config/sources.go
+57 -74
View File
@@ -1,12 +1,13 @@
package main
import (
"context"
"encoding/json"
"errors"
"fmt"
assignerclient "git.novox.be/novox/mesh-contracts/go/clients/meshmoduleassignerclient"
catalogclient "git.novox.be/novox/mesh-delivery/contracts/go/clients/meshcatalogclient"
builds "git.novox.be/novox/mesh-delivery/modules/mesh-delivery-builder/builds"
catalog "git.novox.be/novox/mesh-delivery/contracts/go/seats/meshcatalog"
"sort"
"strings"
)
@@ -91,9 +92,9 @@ func (c *Config) readAssignments(s *Sources) error {
return nil
}
// readDeclarations reads each placed module's declarations from its manifests as the delivery holds them: the trunk's
// (the indexer's `modules.<module>` and its manifest by digest), and, on each node whose want names a build, that
// build's (the builder's `<module>.<fp>` names its manifest digest). On a node the want's build is laid over the
// readDeclarations reads each placed module's declarations from its manifests as the catalog half holds them: the trunk's
// (the manifest its catalog graph entry carries), and, on each node whose want names a build, that build's (the entry
// of its fingerprint). On a node the want's build is laid over the
// trunk's, key by key, so a key only the build it runs declares still resolves there and a key the trunk adds resolves
// before its build is wanted; Declared is the trunk's with every key only a wanted build declares. A placed module
// none of whose manifests can be read is Unresolved, with why.
@@ -186,90 +187,72 @@ func modulesOf(as []Assignment) []string {
return out
}
// trunkDeclarations is a module's declarations at the trunk: the manifest the indexer's `modules.<module>` names.
// trunkDeclarations is a module's declarations at the trunk: the manifest its catalog graph entry carries whole.
func (c *Config) trunkDeclarations(module string) (ModuleDeclarations, error) {
key, err := catalogclient.ModulesKey(module)
e, err := c.trunkEntry(module)
if err != nil {
return ModuleDeclarations{}, err
}
e, err := c.Store.Get(ModulesRead, key)
if err != nil {
return ModuleDeclarations{}, fmt.Errorf("the indexer's modules.%s cannot be read: %w", module, err)
}
if e == nil {
return ModuleDeclarations{}, fmt.Errorf("the indexer holds no modules.%s, so no manifest of it is known", module)
}
var m struct {
Manifest string `json:"manifest"`
}
if err := json.Unmarshal(e.Value, &m); err != nil || m.Manifest == "" {
return ModuleDeclarations{}, fmt.Errorf("the indexer's modules.%s names no manifest", module)
}
return c.manifestDeclarations(module, m.Manifest)
return c.entryDeclarations(e)
}
// builtDeclarations is a module's declarations at a build: the manifest the builder's done key names.
// trunkEntry is a module's entry in the catalog half; with no first view of it, or none listing the module, an error naming
// it, never no declarations.
func (c *Config) trunkEntry(module string) (catalog.CatalogEntry, error) {
e, found, err := c.Catalog.Entry(module)
if err != nil {
return catalog.CatalogEntry{}, fmt.Errorf("the catalog half's entry of %s cannot be read: %w", module, err)
}
if !found {
return catalog.CatalogEntry{}, fmt.Errorf("the catalog half holds no entry of %s, so no manifest of it is known", module)
}
return e, nil
}
// builtDeclarations is a module's declarations at a build: the manifest of the catalog entry of its fingerprint (design 47
// G §594, §614); an entry still asked, or the seat's word that it holds none, is an error naming it.
func (c *Config) builtDeclarations(module, fp string) (ModuleDeclarations, error) {
key, err := builds.BuilderBuildKey(module, fp)
if err != nil {
return ModuleDeclarations{}, err
e, state := c.Catalog.ByFingerprint(context.Background(), fp)
switch state {
case catalogclient.EntryAnswered:
if e.Module != module {
return ModuleDeclarations{}, fmt.Errorf("the catalog entry of %s is %s's, not %s's", short(fp), e.Module, module)
}
return c.entryDeclarations(e)
case catalogclient.EntryAsked:
return ModuleDeclarations{}, fmt.Errorf("the catalog entry of build %s of %s is asked of the seat mesh-catalog and not answered yet", short(fp), module)
}
e, err := c.Store.Get(BuildsRead, key)
if err != nil {
return ModuleDeclarations{}, fmt.Errorf("the builder's %s cannot be read: %w", key, err)
}
if e == nil {
return ModuleDeclarations{}, fmt.Errorf("the builder holds no build %s", key)
}
var b struct {
Manifest string `json:"manifest"`
}
if err := json.Unmarshal(e.Value, &b); err != nil || b.Manifest == "" {
return ModuleDeclarations{}, fmt.Errorf("the builder's %s names no manifest", key)
}
return c.manifestDeclarations(module, b.Manifest)
return ModuleDeclarations{}, fmt.Errorf("the seat mesh-catalog holds no entry of build %s of %s", short(fp), module)
}
// manifestDeclarations is the declarations of the manifest of a digest in the indexer's manifests, read once: a
// manifests key is written once and never changes (ADR 0429 §16), so what it declares is kept by digest.
func (c *Config) manifestDeclarations(module, digest string) (ModuleDeclarations, error) {
c.manifestsMu.Lock()
d, ok := c.manifests[digest]
c.manifestsMu.Unlock()
if ok {
return d, nil
// entryDeclarations is the declarations of an entry's manifest, kept by the manifest's digest where it has one: a manifest
// by its digest never changes, so what it declares is read once.
func (c *Config) entryDeclarations(e catalog.CatalogEntry) (ModuleDeclarations, error) {
digest := e.ManifestDigest
if digest != "" {
c.manifestsMu.Lock()
d, ok := c.manifests[digest]
c.manifestsMu.Unlock()
if ok {
return d, nil
}
}
key, err := catalogclient.ManifestsKey(digest)
raw, err := json.Marshal(e.Manifest)
if err != nil || len(e.Manifest) == 0 {
return ModuleDeclarations{}, fmt.Errorf("the catalog entry of %s carries no manifest", e.Module)
}
d, err := ManifestDeclarations(raw)
if err != nil {
return ModuleDeclarations{}, err
return ModuleDeclarations{}, fmt.Errorf("the manifest of %s: %w", e.Module, err)
}
e, err := c.Store.Get(ManifestsRead, key)
if err != nil {
return ModuleDeclarations{}, fmt.Errorf("the indexer's manifest %s cannot be read: %w", short(digest), err)
if digest != "" {
c.manifestsMu.Lock()
if c.manifests == nil {
c.manifests = map[string]ModuleDeclarations{}
}
c.manifests[digest] = d
c.manifestsMu.Unlock()
}
if e == nil {
return ModuleDeclarations{}, fmt.Errorf("the indexer holds no manifest %s", short(digest))
}
var m struct {
Module string `json:"module"`
JSON string `json:"manifest"`
}
if err := json.Unmarshal(e.Value, &m); err != nil || m.JSON == "" {
return ModuleDeclarations{}, fmt.Errorf("the indexer's manifest %s is not one", short(digest))
}
if m.Module != module {
return ModuleDeclarations{}, fmt.Errorf("the indexer's manifest %s is %s's, not %s's", short(digest), m.Module, module)
}
d, err = ManifestDeclarations([]byte(m.JSON))
if err != nil {
return ModuleDeclarations{}, fmt.Errorf("the manifest %s: %w", short(digest), err)
}
c.manifestsMu.Lock()
if c.manifests == nil {
c.manifests = map[string]ModuleDeclarations{}
}
c.manifests[digest] = d
c.manifestsMu.Unlock()
return d, nil
}
modules/mesh-config/cmd/mesh-config/takeover.go
+5 -17
View File
@@ -13,7 +13,6 @@ import (
"encoding/json"
"errors"
"fmt"
catalogclient "git.novox.be/novox/mesh-delivery/contracts/go/clients/meshcatalogclient"
catalog "git.novox.be/novox/mesh-delivery/contracts/go/seats/meshcatalog"
"time"
)
@@ -44,8 +43,8 @@ func (c *Config) source() (Switch, error) {
// ---- declared --------------------------------------------------------------------------------------------
// DeclaredKept is one key of `declared`: a module's declarations as the manifest the indexer's `modules.<module>`
// names holds them, with that manifest's digest and commit (ADR 0565 §3).
// DeclaredKept is one key of `declared`: a module's declarations as the manifest its catalog graph entry
// carries holds them, with that manifest's digest and commit (ADR 0565 §3).
type DeclaredKept struct {
Module string `json:"module"`
Manifest string `json:"manifest"`
@@ -55,26 +54,15 @@ type DeclaredKept struct {
// trunkKept is a module's declarations at the trunk with the manifest they come from.
func (c *Config) trunkKept(module string) (DeclaredKept, error) {
key, err := catalogclient.ModulesKey(module)
e, err := c.trunkEntry(module)
if err != nil {
return DeclaredKept{}, err
}
e, err := c.Store.Get(ModulesRead, key)
if err != nil || e == nil {
return DeclaredKept{}, fmt.Errorf("the indexer's modules.%s cannot be read (%v)", module, err)
}
var m struct {
Manifest string `json:"manifest"`
Commit string `json:"commit"`
}
if err := json.Unmarshal(e.Value, &m); err != nil || m.Manifest == "" {
return DeclaredKept{}, fmt.Errorf("the indexer's modules.%s names no manifest", module)
}
d, err := c.manifestDeclarations(module, m.Manifest)
d, err := c.entryDeclarations(e)
if err != nil {
return DeclaredKept{}, err
}
return DeclaredKept{Module: module, Manifest: m.Manifest, Commit: m.Commit, Config: d.Config}, nil
return DeclaredKept{Module: module, Manifest: e.ManifestDigest, Commit: e.Commit, Config: d.Config}, nil
}
// writeDeclared makes `declared` hold each placed module's trunk declarations, writing a key only when it differs
modules/mesh-config/cmd/mesh-config/whole_integration_test.go
+76 -28
View File
@@ -24,8 +24,9 @@ import (
nodeengine "git.novox.be/novox/mesh-contracts/go/seats/nodeengine"
catalogclient "git.novox.be/novox/mesh-delivery/contracts/go/clients/meshcatalogclient"
catalog "git.novox.be/novox/mesh-delivery/contracts/go/seats/meshcatalog"
builds "git.novox.be/novox/mesh-delivery/modules/mesh-delivery-builder/builds"
delivery "git.novox.be/novox/mesh-delivery/contracts/go/seats/meshdelivery"
"io"
"maps"
"os"
"os/exec"
"path/filepath"
@@ -74,6 +75,11 @@ type player struct {
subs []string
// owned is each delivery owner's state as the owner's tool answers it: its values and the revision of its last change.
owned map[string]*ownedState
// catalog is the seat mesh-catalog's trunk revision of the watched branch main, as its verb revision answers it: each
// module's entry, at the revision catRev (0: none yet); byFp is the entry of a build's fingerprint, as entry answers it.
catalog map[string]catalog.CatalogEntry
byFp map[string]catalog.CatalogEntry
catRev int64
}
// ownedState is one state of an owner (the indexer's modules, the assigner's wants...), kept from its bucket.
@@ -364,6 +370,26 @@ func (p *player) asked(key string, body json.RawMessage) any {
return a
}
switch {
case key == catalogclient.RegistrationsVerb:
return catalog.RegistrationsAnswer{Registrations: []catalog.Registration{{Module: Module, Version: 1, Branches: []string{"main"}}}}
case key == catalogclient.SourceVerb:
return delivery.SourceAnswer{}
case key == catalogclient.RevisionVerb:
if p.catRev == 0 {
return catalog.RevisionAnswer{NoneYet: true}
}
entries := make([]catalog.CatalogEntry, 0, len(p.catalog))
for _, name := range slices.Sorted(maps.Keys(p.catalog)) {
entries = append(entries, p.catalog[name])
}
return catalog.RevisionAnswer{Revision: &catalog.CatalogRevision{Revision: p.catRev, Kind: catalogclient.KindTrunk, Branch: "main"}, Entries: entries}
case key == catalogclient.EntryVerb:
var q catalog.EntryInput
_ = json.Unmarshal(body, &q)
if e, ok := p.byFp[q.Fingerprint]; ok {
return catalog.EntryAnswer{Entry: e}
}
return map[string]any{"error": "no entry of " + q.Fingerprint}
case key == HavesAsk:
var q struct{ Node, Module string }
_ = json.Unmarshal(body, &q)
@@ -543,7 +569,7 @@ func (p *player) viewsReady() {
Answered time.Time `json:"Answered"`
} `json:"first-views"`
}
if json.Unmarshal(raw, &a) != nil || len(a.Views) != len(followed) {
if json.Unmarshal(raw, &a) != nil || len(a.Views) != len(followed)+2 { // the registrations and the branch main
return false
}
for _, v := range a.Views {
@@ -565,30 +591,52 @@ func eventually(t *testing.T, what string, ok func() bool) {
t.Fatalf("never: %s", what)
}
// manifestIn writes a module's manifest into the indexer's buckets as the indexer writes it, and names it as the trunk's.
func (p *player) manifestIn(module, raw string) string {
// fpOf is a module's build at a base fingerprint: a fingerprint names one entry of the seat mesh-catalog, so two modules never
// share one.
func fpOf(module, base string) string { return base + ":" + module }
// manifestIn makes a module's manifest the trunk's: the seat mesh-catalog's next revision lists its entry, and says it as
// the event catalog-revision, as the seat does.
func (p *player) manifestIn(module, raw string) {
p.t.Helper()
digest := catalogclient.ManifestDigest([]byte(raw))
value, err := catalogclient.EncodeManifest(catalog.ManifestsByDigestValue{Module: module, Manifest: raw})
if err != nil {
var whole map[string]any
if err := json.Unmarshal([]byte(raw), &whole); err != nil {
p.t.Fatal(err)
}
p.put("mesh-delivery-indexer.manifests", digest, json.RawMessage(value))
p.put("mesh-delivery-indexer.modules", module, map[string]any{"v": 1, "manifest": digest, "repository": "novox/" + module, "commit": "c1"})
return digest
p.mu.Lock()
p.catalog[module] = catalog.CatalogEntry{Module: module, Manifest: whole, ManifestDigest: catalogclient.ManifestDigest([]byte(raw)), Commit: "c1"}
p.catRev++
rev := p.catRev
p.mu.Unlock()
p.hearsCatalog()
// The seat's event on the branch's part, as the bus says it (design 47 G §532): the module's subscription is `<seat>.<event>.>`.
p.event(catalog.Seat+"."+catalog.EventCatalogRevision+".main", catalog.CatalogRevisionEvent{Revision: rev, Kind: catalogclient.KindTrunk, Branch: "main", Changed: []string{module}})
}
// buildIn writes a build of a module with a manifest, and the want that names it on a node.
// buildIn makes the seat mesh-catalog hold the entry of a build's fingerprint, and the want that names it on a node.
func (p *player) buildIn(node, module, fp, raw string) {
p.t.Helper()
digest := catalogclient.ManifestDigest([]byte(raw))
value, _ := catalogclient.EncodeManifest(catalog.ManifestsByDigestValue{Module: module, Manifest: raw})
p.put("mesh-delivery-indexer.manifests", digest, json.RawMessage(value))
bk, _ := builds.BuilderBuildKey(module, fp)
p.put("mesh-delivery-builder.builds", bk, map[string]any{"v": 1, "state": "done", "manifest": digest})
var whole map[string]any
if err := json.Unmarshal([]byte(raw), &whole); err != nil {
p.t.Fatal(err)
}
p.mu.Lock()
p.byFp[fp] = catalog.CatalogEntry{Module: module, Fp: fp, Manifest: whole, ManifestDigest: catalogclient.ManifestDigest([]byte(raw))}
p.mu.Unlock()
p.put("mesh-delivery-assigner.wants", "want."+node+"."+module, map[string]any{"v": 1, "fp": fp, "self": true})
}
// hearsCatalog waits until the module subscribed to the seat's event catalog-revision: a revision said before that would
// reach it only as the first view it asks for. Waited on, never slept for.
func (p *player) hearsCatalog() {
p.t.Helper()
eventually(p.t, "the module subscribed to catalog-revision", func() bool {
p.mu.Lock()
defer p.mu.Unlock()
return slices.Contains(p.subs, catalog.Seat+"."+catalog.EventCatalogRevision+".>")
})
}
func (p *player) placed(node string, modules ...string) {
p.t.Helper()
pl := meshseat.Assignments{V: 1, Node: node}
@@ -626,7 +674,7 @@ func start(t *testing.T, bin string, js jetstream.JetStream, router *standIn, ha
stdout, _ := cmd.StdoutPipe()
cmd.Stderr = os.Stderr
p := &player{t: t, ctx: ctx, js: js, router: router, in: stdin, pending: map[string]chan wire{}, haves: haves, readers: readers,
owned: map[string]*ownedState{}}
owned: map[string]*ownedState{}, catalog: map[string]catalog.CatalogEntry{}, byFp: map[string]catalog.CatalogEntry{}}
for _, f := range followed {
p.own(f.Read)
}
@@ -658,8 +706,8 @@ func TestConfigWholeRenamesAndRetiresAgainstTheBus(t *testing.T) {
if err != nil {
t.Fatal(err)
}
buckets := []string{"mesh-delivery-assigner_assignments", "mesh-delivery-assigner_wants", "mesh-delivery-builder_builds",
"mesh-delivery-indexer_modules", "mesh-delivery-indexer_manifests", "mesh-delivery-indexer_predefined",
buckets := []string{"mesh-delivery-assigner_assignments", "mesh-delivery-assigner_wants",
"mesh-delivery-indexer_predefined",
"mesh-delivery-indexer_config-candidates"}
for _, s := range []string{ValuesState, ResolvedState, PredefinedState, ProposalsState, PullsState, KeptState, DeclaredState,
RenamesState, RetiringState, FindingsState} {
@@ -671,7 +719,7 @@ func TestConfigWholeRenamesAndRetiresAgainstTheBus(t *testing.T) {
}
}
router := newStandIn()
haves := map[string]string{"g14.player": fp1, "novox.player": fp1, "g14.radio": fp1}
haves := map[string]string{"g14.player": fpOf("player", fp1), "novox.player": fpOf("player", fp1), "g14.radio": fpOf("radio", fp1)}
p := start(t, bin, js, router, haves, nil)
p.viewsReady()
@@ -681,9 +729,9 @@ func TestConfigWholeRenamesAndRetiresAgainstTheBus(t *testing.T) {
const dock = `{"module":"dock","config":{"a":{"type":"number","default":1},"b":{"type":"number","default":1,"renamed-from":"a"}}}`
p.manifestIn("player", oldPlayer)
p.manifestIn("radio", oldRadio)
p.buildIn("g14", "player", fp1, oldPlayer)
p.buildIn("novox", "player", fp1, oldPlayer)
p.buildIn("g14", "radio", fp1, oldRadio)
p.buildIn("g14", "player", fpOf("player", fp1), oldPlayer)
p.buildIn("novox", "player", fpOf("player", fp1), oldPlayer)
p.buildIn("g14", "radio", fpOf("radio", fp1), oldRadio)
p.placed("g14", "player", "radio")
p.placed("novox", "player")
if _, err := p.call(ResyncTool, nil); err != nil {
@@ -720,7 +768,7 @@ func TestConfigWholeRenamesAndRetiresAgainstTheBus(t *testing.T) {
// A refused move writes its finding: dock declares both names.
p.manifestIn("dock", dock)
p.buildIn("g14", "dock", fp1, dock)
p.buildIn("g14", "dock", fpOf("dock", fp1), dock)
p.placed("g14", "player", "radio", "dock")
eventually(t, "the refused rename's finding", func() bool {
keys, _ := (func() ([]string, error) { kv, _ := p.kv(FindingsState); return kv.Keys(p.ctx) })()
@@ -768,9 +816,9 @@ func TestConfigWholeRenamesAndRetiresAgainstTheBus(t *testing.T) {
}
// g14 applies the build that reads the new name; the delivery says so; its window closes, novox's stays.
p2.buildIn("g14", "player", fp2, newPlayer)
p2.buildIn("g14", "player", fpOf("player", fp2), newPlayer)
p2.mu.Lock()
p2.haves["g14.player"] = fp2
p2.haves["g14.player"] = fpOf("player", fp2)
p2.mu.Unlock()
p2.event("mesh-delivery-view.delivered", map[string]any{"revision": "r1"})
eventually(t, "g14's window closes and the old name goes", func() bool {
@@ -792,9 +840,9 @@ func TestConfigWholeRenamesAndRetiresAgainstTheBus(t *testing.T) {
if len(router.asks) != 0 || p2.publishedCount(EventRetiring) != 0 {
t.Fatalf("asked %d times, retiring said %d times while g14 reads the value", len(router.asks), p2.publishedCount(EventRetiring))
}
p2.buildIn("g14", "radio", fp2, newRadio)
p2.buildIn("g14", "radio", fpOf("radio", fp2), newRadio)
p2.mu.Lock()
p2.haves["g14.radio"] = fp2
p2.haves["g14.radio"] = fpOf("radio", fp2)
p2.mu.Unlock()
p2.event("mesh-delivery-view.delivered", map[string]any{"revision": "r2"})
eventually(t, "the operator is asked once", func() bool {
modules/mesh-config/go.mod
-1
View File
@@ -16,7 +16,6 @@ require (
git.novox.be/novox/mesh-delivery/contracts/go/clients/meshcatalogclient v0.0.0-00010101000000-000000000000
git.novox.be/novox/mesh-delivery/contracts/go/seats/meshcatalog v0.0.0-00010101000000-000000000000
git.novox.be/novox/mesh-delivery/contracts/go/seats/meshwants v0.0.0-00010101000000-000000000000
git.novox.be/novox/mesh-delivery/modules/mesh-delivery-builder/builds v0.0.0-00010101000000-000000000000
git.novox.be/novox/mesh-sdk/go v0.0.0-00010101000000-000000000000
github.com/nats-io/nats.go v1.54.0
)
modules/mesh-config/module.json
+8 -8
View File
@@ -28,12 +28,13 @@
"mesh-delivery-view.delivered",
"node-engine.manifest-reads-changed",
"mesh-catalog.manifest-accepted",
"mesh-delivery-indexer.views.modules",
"mesh-delivery-indexer.views.manifests",
"mesh-delivery-indexer.views.predefined",
"mesh-delivery-indexer.views.config-candidates",
"mesh-delivery-assigner.views.assignments",
"mesh-delivery-assigner.views.wants",
"mesh-catalog.catalog-revision",
"mesh-catalog.registration-changed",
"mesh-delivery.source-moved",
"mesh-conditions.findings-asked"
],
"invokes": [
@@ -44,15 +45,14 @@
"seat:node-engine.manifest-reads",
"mesh-delivery-indexer.indexer_config_readers",
"seat:mesh-vault.retire",
"mesh-delivery-indexer.indexer_read_modules",
"mesh-delivery-indexer.indexer_read_manifests",
"mesh-delivery-indexer.indexer_read_predefined",
"mesh-delivery-indexer.indexer_read_config_candidates",
"mesh-delivery-assigner.assigner_read_assignments",
"mesh-delivery-assigner.assigner_read_wants"
],
"reads": [
"mesh-delivery-builder.builds"
"mesh-delivery-assigner.assigner_read_wants",
"seat:mesh-catalog.revision",
"seat:mesh-catalog.registrations",
"seat:mesh-catalog.entry",
"seat:mesh-delivery.source"
],
"tools": [
"config_declared",
modules/mesh-provisioner/cmd/mesh-provisioner/catalogfake_test.go
+79
View File
@@ -0,0 +1,79 @@
package main
import (
"context"
"encoding/json"
"sync"
catalogclient "git.novox.be/novox/mesh-delivery/contracts/go/clients/meshcatalogclient"
catalog "git.novox.be/novox/mesh-delivery/contracts/go/seats/meshcatalog"
)
// fakeCatalog is the catalog half as the tests give it: each module's trunk entry, and the entries a fingerprint names.
type fakeCatalog struct {
mu sync.Mutex
entries map[string]catalog.CatalogEntry
byFp map[string]catalog.CatalogEntry
notYet error
}
func newFakeCatalog() *fakeCatalog {
return &fakeCatalog{entries: map[string]catalog.CatalogEntry{}, byFp: map[string]catalog.CatalogEntry{}}
}
// manifestMap is a manifest's JSON as the entry carries it whole.
func manifestMap(raw string) map[string]any {
var m map[string]any
if err := json.Unmarshal([]byte(raw), &m); err != nil {
panic(err)
}
return m
}
// set gives a module's trunk entry, its manifest the JSON raw and its fingerprint fp.
func (f *fakeCatalog) set(module, raw, fp string) {
f.mu.Lock()
defer f.mu.Unlock()
f.entries[module] = catalog.CatalogEntry{Module: module, Fp: fp, Manifest: manifestMap(raw)}
}
// setFp changes the fingerprint of a module's trunk entry.
func (f *fakeCatalog) setFp(module, fp string) {
f.mu.Lock()
defer f.mu.Unlock()
e := f.entries[module]
e.Fp = fp
f.entries[module] = e
}
// setBuild gives the entry the fingerprint of an older build names.
func (f *fakeCatalog) setBuild(module, fp, raw string) {
f.mu.Lock()
defer f.mu.Unlock()
f.byFp[fp] = catalog.CatalogEntry{Module: module, Fp: fp, Manifest: manifestMap(raw)}
}
func (f *fakeCatalog) Entry(module string) (catalog.CatalogEntry, bool, error) {
f.mu.Lock()
defer f.mu.Unlock()
if f.notYet != nil {
return catalog.CatalogEntry{}, false, f.notYet
}
e, ok := f.entries[module]
return e, ok, nil
}
func (f *fakeCatalog) Ready() error {
f.mu.Lock()
defer f.mu.Unlock()
return f.notYet
}
func (f *fakeCatalog) ByFingerprint(_ context.Context, fp string) (catalog.CatalogEntry, catalogclient.EntryState) {
f.mu.Lock()
defer f.mu.Unlock()
if e, ok := f.byFp[fp]; ok {
return e, catalogclient.EntryAnswered
}
return catalog.CatalogEntry{}, catalogclient.EntryNotFound
}
modules/mesh-provisioner/cmd/mesh-provisioner/catalogfeed.go
+68
View File
@@ -0,0 +1,68 @@
package main
// The provisioner's reading of which modules exist and of a provider's manifest (design 47 G §196, §594, §614): the catalog
// half, taken through meshcatalogclient's follower of the watched branches on the part's one views reader. A module's
// manifest is the one its catalog graph entry carries whole (§443), never the indexer's state `modules` or `manifests`; the
// manifest a running build was made from is the entry read by that build's fingerprint (§614, an older build's entry).
import (
"context"
"fmt"
"strings"
"git.novox.be/novox/mesh-delivery/contracts/go/clients/meshcatalogclient"
catalog "git.novox.be/novox/mesh-delivery/contracts/go/seats/meshcatalog"
)
// CatalogHalf is what the holder reads of the catalog half.
type CatalogHalf interface {
// Entry is a module's entry at the trunk; an error while a watched branch has no first view yet, never "no module".
Entry(module string) (catalog.CatalogEntry, bool, error)
// Ready is nil once every watched branch has its first view (none-yet counting as one), else says what it waits for.
Ready() error
// ByFingerprint reads an entry by the fingerprint of a build: from a branch's half when one lists it, else asked of the
// seat's verb `entry` once and kept (asked, answered or not found).
ByFingerprint(ctx context.Context, fingerprint string) (catalog.CatalogEntry, meshcatalogclient.EntryState)
}
// branchesHalf is the CatalogHalf over the follower of the watched branches.
type branchesHalf struct{ b *meshcatalogclient.Branches }
func (h branchesHalf) Ready() error {
followed := h.b.Followed()
if len(followed) == 0 {
return fmt.Errorf("the catalog half names no watched branch yet (the seat mesh-catalog's registrations are asked until they answer)")
}
var waiting []string
for _, branch := range followed {
if c, ok := h.b.Catalog(branch); ok {
select {
case <-c.Ready():
default:
waiting = append(waiting, branch)
}
}
}
if len(waiting) > 0 {
return fmt.Errorf("the catalog half has no first view of %s yet", strings.Join(waiting, ", "))
}
return nil
}
func (h branchesHalf) Entry(module string) (catalog.CatalogEntry, bool, error) {
if err := h.Ready(); err != nil {
return catalog.CatalogEntry{}, false, err
}
for _, branch := range h.b.Followed() { // the branch followed first names a module more than one hold
if c, ok := h.b.Catalog(branch); ok {
if e, ok := c.Entry(module); ok {
return e, true, nil
}
}
}
return catalog.CatalogEntry{}, false, nil
}
func (h branchesHalf) ByFingerprint(ctx context.Context, fingerprint string) (catalog.CatalogEntry, meshcatalogclient.EntryState) {
return h.b.Resolve(ctx, fingerprint)
}
modules/mesh-provisioner/cmd/mesh-provisioner/catalogfeed_test.go
+90
View File
@@ -0,0 +1,90 @@
package main
import (
"context"
"encoding/json"
"fmt"
"strings"
"sync"
"testing"
"time"
"git.novox.be/novox/mesh-delivery/contracts/go/clients/meshcatalogclient"
catalog "git.novox.be/novox/mesh-delivery/contracts/go/seats/meshcatalog"
delivery "git.novox.be/novox/mesh-delivery/contracts/go/seats/meshdelivery"
"git.novox.be/novox/mesh-sdk/go/views"
)
// wiringOf is a part's reader in miniature: it keeps the followers adopted, so a test starts them as the reader's Start does.
type wiringOf struct {
mu sync.Mutex
fs []*views.Follower
}
func (w *wiringOf) Adopt(f *views.Follower, _, _ string, _ func(context.Context, []byte) error) error {
w.mu.Lock()
defer w.mu.Unlock()
w.fs = append(w.fs, f)
return nil
}
func (w *wiringOf) Drop(string, string) error { return nil }
// Design 47 G §196, §594: which modules exist, and the manifest of each, are the catalog half's; a branch with no first
// view yet is an error naming it, never no module, and an entry's manifest is the one it carries whole.
func TestCatalogHalfIsTheWatchedBranchesNewestRevision(t *testing.T) {
ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
defer cancel()
answered := false
var mu sync.Mutex
call := func(key string, _ any) (json.RawMessage, error) {
mu.Lock()
defer mu.Unlock()
switch key {
case meshcatalogclient.RegistrationsVerb:
return json.Marshal(catalog.RegistrationsAnswer{Registrations: []catalog.Registration{{Module: "m", Version: 1, Branches: []string{"main"}}}})
case meshcatalogclient.SourceVerb:
return json.Marshal(delivery.SourceAnswer{})
case meshcatalogclient.RevisionVerb:
if !answered {
return nil, fmt.Errorf("the holder is away")
}
return json.Marshal(catalog.RevisionAnswer{Revision: &catalog.CatalogRevision{Revision: 1, Kind: meshcatalogclient.KindTrunk},
Entries: []catalog.CatalogEntry{{Module: "pg", Manifest: map[string]any{"module": "pg"}}}})
}
return nil, fmt.Errorf("asked %s", key)
}
w := &wiringOf{}
b, err := meshcatalogclient.NewBranches(w, call)
if err != nil {
t.Fatal(err)
}
h := branchesHalf{b}
if _, _, err := h.Entry("pg"); err == nil || !strings.Contains(err.Error(), "no watched branch") {
t.Fatalf("before the registrations: %v, want an error naming the missing branch", err)
}
go func() { _ = w.fs[0].Start(ctx) }() // the registrations
for len(b.Followed()) == 0 {
select {
case <-ctx.Done():
t.Fatal("the registrations' branch is never followed")
case <-time.After(10 * time.Millisecond):
}
}
if _, _, err := h.Entry("pg"); err == nil || !strings.Contains(err.Error(), "no first view of main") {
t.Fatalf("with no first view: %v, want an error naming the branch, never no module", err)
}
mu.Lock()
answered = true
mu.Unlock()
c, _ := b.Catalog("main")
if err := c.Start(ctx); err != nil {
t.Fatal(err)
}
got, ok, err := h.Entry("pg")
if err != nil || !ok || got.Manifest["module"] != "pg" || h.Ready() != nil {
t.Fatalf("with the first view: %v, %v, %v", got, ok, err)
}
if _, ok, err := h.Entry("absent"); ok || err != nil {
t.Fatalf("a module the half does not list: %v, %v", ok, err)
}
}
modules/mesh-provisioner/cmd/mesh-provisioner/composed_test.go
+10 -22
View File
@@ -15,9 +15,7 @@ import (
assignerclient "git.novox.be/novox/mesh-contracts/go/clients/meshmoduleassignerclient"
engineclient "git.novox.be/novox/mesh-contracts/go/clients/nodeengineclient"
meshseat "git.novox.be/novox/mesh-contracts/go/seats/meshmoduleassigner"
catalogclient "git.novox.be/novox/mesh-delivery/contracts/go/clients/meshcatalogclient"
"git.novox.be/novox/mesh-delivery/contracts/go/clients/meshcatalogkey"
catalog "git.novox.be/novox/mesh-delivery/contracts/go/seats/meshcatalog"
"os"
"slices"
"strings"
@@ -83,7 +81,7 @@ func writeAssignments(t *testing.T, s *memStore, node string, modules ...string)
}
}
// goldenMesh is the store as the provisioner reads it for frombustest's mesh: the indexer's modules, n1's assignments, the
// goldenMesh is the store as the provisioner reads it for frombustest's mesh: the catalog half's modules, n1's assignments, the
// vault's on n2, restic's want.
func goldenMesh(t *testing.T, c today) *memStore {
t.Helper()
@@ -95,10 +93,7 @@ func goldenMesh(t *testing.T, c today) *memStore {
raw = `{"module":"postgres","version":"1","provides":[{"name":"postgres-database","scope":"mesh"}],` +
`"serves":{"postgres-database":{"port":5432}}}`
}
digest := catalogclient.ManifestDigest([]byte(raw))
value, _ := catalogclient.EncodeManifest(catalog.ManifestsByDigestValue{Module: module, Manifest: raw})
s.write(t, ManifestsRead, mustKey(catalogclient.ManifestsKey(digest)), string(value))
s.write(t, ModulesRead, module, `{"v":1,"fp2":"`+fp+`","manifest":"`+digest+`"}`)
s.cat.set(module, raw, fp)
}
writeAssignments(t, s, c.Node, "store", "postgres", "mesh-provisioner-client", "restic")
writeAssignments(t, s, "n2", "mesh-vault")
@@ -118,8 +113,8 @@ func TestThePlansFromCompositionsAreTodays(t *testing.T) {
s := goldenMesh(t, c)
published(t, s, c)
now := time.Date(2026, 11, 2, 18, 0, 0, 0, time.UTC)
h := &Holder{Store: s, Asks: &router{}, Logf: t.Logf, Now: func() time.Time { return now }}
h.Compose = FromCompositions(s)
h := &Holder{Store: s, Catalog: s.cat, Asks: &router{}, Logf: t.Logf, Now: func() time.Time { return now }}
h.Compose = FromCompositions(s, s.cat)
out, err := h.ComposeNode(context.Background(), c.Node)
if err != nil {
t.Fatal(err)
@@ -155,8 +150,8 @@ func TestARefusedOrMissingCompositionHoldsItsModule(t *testing.T) {
readJSON(t, "testdata/today-composed.json", &c)
s := goldenMesh(t, c)
published(t, s, c)
h := &Holder{Store: s, Asks: &router{}, Logf: t.Logf, Now: time.Now}
h.Compose = FromCompositions(s)
h := &Holder{Store: s, Catalog: s.cat, Asks: &router{}, Logf: t.Logf, Now: time.Now}
h.Compose = FromCompositions(s, s.cat)
if _, err := h.ComposeNode(context.Background(), c.Node); err != nil {
t.Fatal(err)
}
@@ -207,12 +202,11 @@ func TestARefusedOrMissingCompositionHoldsItsModule(t *testing.T) {
{"refused", `{"v":1,"node":"n1","module":"restic","fp":"` + fpsOf["restic"] + `","resources":[],"refused":"no postgres here"}`, "no postgres here"},
} {
ahead := "src2:" + strings.Repeat("7", 64)
e, _ := s.Get(ModulesRead, "restic")
s.write(t, ModulesRead, "restic", strings.ReplaceAll(string(e.Value), fpsOf["restic"], ahead))
s.cat.setFp("restic", ahead)
if row.composed != "" {
s.write(t, ComposedRead, mustKey(composedKey("n1", "restic", ahead)), row.composed)
}
comp := FromCompositions(s)(context.Background(), c.Node, map[string]string{"restic": fpsOf["restic"]})
comp := FromCompositions(s, s.cat)(context.Background(), c.Node, map[string]string{"restic": fpsOf["restic"]})
if !strings.Contains(comp.Newest.Held["restic"], row.held) || slices.Contains(comp.UnpublishedNewest, "restic") != (row.composed == "") {
t.Errorf("%s: newest held %v, unpublished at the newest %v", row.name, comp.Newest.Held, comp.UnpublishedNewest)
}
@@ -224,12 +218,12 @@ func TestARefusedOrMissingCompositionHoldsItsModule(t *testing.T) {
if e, _ := s.Get(RefusedState, "n1.restic=newest"); e == nil || !strings.Contains(string(e.Value), row.held) {
t.Errorf("%s: refused holds %s", row.name, valueOf(e))
}
h.Compose = FromCompositions(s)
h.Compose = FromCompositions(s, s.cat)
if _, held := comp.AtWants.Held["restic"]; held {
t.Errorf("%s: its wanted build is composed, but the module is held there: %v", row.name, comp.AtWants.Held)
}
}
s.write(t, ModulesRead, "restic", strings.ReplaceAll(valueOf(mustEntry(t, s)), "src2:"+strings.Repeat("7", 64), fpsOf["restic"]))
s.cat.setFp("restic", fpsOf["restic"])
if _, err := h.ComposeNode(context.Background(), c.Node); err != nil {
t.Fatal(err)
}
@@ -238,12 +232,6 @@ func TestARefusedOrMissingCompositionHoldsItsModule(t *testing.T) {
}
}
func mustEntry(t *testing.T, s *memStore) *Entry {
t.Helper()
e, _ := s.Get(ModulesRead, "restic")
return e
}
func readJSON(t *testing.T, path string, v any) {
t.Helper()
raw, err := os.ReadFile(path)
modules/mesh-provisioner/cmd/mesh-provisioner/holder.go
+27 -65
View File
@@ -36,8 +36,6 @@ import (
meshseat "git.novox.be/novox/mesh-contracts/go/seats/meshmoduleassigner"
catalogclient "git.novox.be/novox/mesh-delivery/contracts/go/clients/meshcatalogclient"
"git.novox.be/novox/mesh-delivery/contracts/go/clients/meshcatalogkey"
catalog "git.novox.be/novox/mesh-delivery/contracts/go/seats/meshcatalog"
builds "git.novox.be/novox/mesh-delivery/modules/mesh-delivery-builder/builds"
"slices"
"sort"
"strings"
@@ -75,13 +73,8 @@ type Refusal struct {
const (
AssignmentsRead = (assignerModule + "." + meshseat.StateAssignments)
WantsRead = assignerModule + ".wants"
ModulesRead = indexerModule + "." + catalog.StateModules
ManifestsRead = indexerModule + "." + catalog.StateManifests
HaveRead = "node-engine.have"
// BuildsRead is the builder's builds, `<module>.<fingerprint part>`: the key of the build a provider runs names the
// digest of the manifest it was built from (novox/hq ADR 0603 clause 5).
BuildsRead = builds.BuilderModule + ".builds"
ComposedRead = "mesh-host.composed"
ComposedRead = "mesh-host.composed"
)
// BoundReads are the providers' entries this module's manifest reads, `<module>.bound`, whose changes it hears. Which
@@ -142,7 +135,9 @@ type Asker interface {
// Holder is the seat's holder.
type Holder struct {
Store Store
Store Store
// Catalog is the catalog half: which modules exist and the manifest each entry carries (design 47 G §594).
Catalog CatalogHalf
Compose Composer
Asks Asker
Now func() time.Time
@@ -904,38 +899,22 @@ func (h *Holder) haveOutcome(p *meshprovisionerclient.Provision) {
}
}
// buildOffers says whether the build of a module at a fingerprint, as the builder keyed it, was made from a manifest that
// offers the provision; known is false when the builder holds no such key, the key names no manifest, or the indexer holds
// no such manifest (the caller then does not refuse).
// buildOffers says whether the build of a module at a fingerprint was made from a manifest that offers the provision,
// read from the catalog entry of that fingerprint (design 47 G §594, §614); known is false while the entry is asked or when
// the seat holds none (the caller then does not refuse).
func (h *Holder) buildOffers(module, fingerprint, provision string) (offers, known bool) {
key, err := builds.BuilderBuildKey(module, fingerprint)
if err != nil {
entry, state := h.Catalog.ByFingerprint(context.Background(), fingerprint)
if state != catalogclient.EntryAnswered || entry.Module != module {
return false, false
}
be, err := h.Store.Get(BuildsRead, key)
if err != nil || be == nil {
return false, false
}
b, err := builds.DecodeBuilderBuild(be.Value)
if err != nil || b.Manifest == "" {
return false, false
}
mk, err := catalogclient.ManifestsKey(b.Manifest)
if err != nil {
return false, false
}
me, err := h.Store.Get(ManifestsRead, mk)
if err != nil || me == nil {
return false, false
}
man, err := catalogclient.DecodeManifest(me.Value)
raw, err := json.Marshal(entry.Manifest)
if err != nil {
return false, false
}
var decl struct {
Provides []json.RawMessage `json:"provides"`
}
if json.Unmarshal([]byte(man.Manifest), &decl) != nil {
if json.Unmarshal(raw, &decl) != nil {
return false, false
}
for _, raw := range decl.Provides {
@@ -1011,33 +990,20 @@ func keepsEntriesFor(manifest, provision string) (bool, error) {
return helped || !says, nil
}
// providerManifest is a provider's manifest JSON at the trunk, as the indexer holds it.
// providerManifest is a provider's manifest JSON at the trunk, as its catalog graph entry carries it whole.
func (h *Holder) providerManifest(module string) (string, error) {
mk, err := catalogclient.ModulesKey(module)
e, ok, err := h.Catalog.Entry(module)
if err != nil {
return "", err
}
e, err := h.Store.Get(ModulesRead, mk)
if err != nil || e == nil {
return "", fmt.Errorf("the indexer holds no modules.%s (%v)", module, err)
if !ok || len(e.Manifest) == 0 {
return "", fmt.Errorf("the catalog half holds no manifest of %s", module)
}
m, err := catalogclient.DecodeModule(e.Value)
if err != nil || m.Manifest == "" {
return "", fmt.Errorf("the indexer's modules.%s names no manifest", module)
}
key, err := catalogclient.ManifestsKey(m.Manifest)
raw, err := json.Marshal(e.Manifest)
if err != nil {
return "", err
}
me, err := h.Store.Get(ManifestsRead, key)
if err != nil || me == nil {
return "", fmt.Errorf("the indexer holds no manifest of %s (%v)", module, err)
}
man, err := catalogclient.DecodeManifest(me.Value)
if err != nil {
return "", err
}
return man.Manifest, nil
return string(raw), nil
}
// retired says a retiring key the operator approved is retired: its provider's entry is gone.
@@ -1544,13 +1510,13 @@ func orSaid(words ...string) string {
// FromCompositions is the holder's composer over the node-engine's published compositions (ADR 0593 §8): it composes
// nothing, and takes each module's plan from the key of `composed-<node>` at the build it is read at. At the wanted
// builds a module is read at the fingerprint keptAt names, else at the indexer's newest (FP2); at the newest, each placed
// builds a module is read at the fingerprint keptAt names, else at the catalog entry's fingerprint (FP2); at the newest, each placed
// module at FP2. A key whose engine refused the module holds it with the engine's words; a key not published yet holds
// it at its wanted build (its last plan kept, as for a refusal) and at its newest too (issue 1966), whose build the engine
// writes once it is done (ADR 0593 §6), and says it unpublished, so the pass judges no retirement on the node (issue
// 940). A placed module no build of which is known is held too. A plan absent from a key is a module that hands the
// provisioner nothing.
func FromCompositions(s Store) Composer {
func FromCompositions(s Store, cat CatalogHalf) Composer {
return func(_ context.Context, node string, keptAt map[string]string) Composition {
start := time.Now()
at, newest := HeldPlans{Plans: map[string]catalogue.Plan{}, Held: map[string]string{}},
@@ -1560,7 +1526,7 @@ func FromCompositions(s Store) Composer {
at.Err, newest.Err = err, err
return Composition{AtWants: at, Newest: newest, Took: []Phase{{Name: "compositions read", Took: time.Since(start)}}}
}
placed, fp2, err := placedModules(s, node)
placed, fp2, err := placedModules(s, cat, node)
if err != nil {
at.Err, newest.Err = err, err
return Composition{AtWants: at, Newest: newest, Took: []Phase{{Name: "compositions read", Took: time.Since(start)}}}
@@ -1662,7 +1628,7 @@ func publishedOn(s Store, node string) (map[string]publishedPlan, error) {
}
// placedModules is every module placed on node, sorted, and the indexer's newest fingerprint (FP2) of each.
func placedModules(s Store, node string) ([]string, map[string]string, error) {
func placedModules(s Store, cat CatalogHalf, node string) ([]string, map[string]string, error) {
raw := map[string][]byte{}
keys, err := s.Keys(AssignmentsRead)
if err != nil {
@@ -1681,14 +1647,12 @@ func placedModules(s Store, node string) ([]string, map[string]string, error) {
fp2 := map[string]string{}
for _, m := range assignments[node].Modules {
placed = append(placed, m.Module)
mk, err := catalogclient.ModulesKey(m.Module)
e, ok, err := cat.Entry(m.Module)
if err != nil {
continue
return nil, nil, err
}
if e, err := s.Get(ModulesRead, mk); err == nil && e != nil {
if mod, err := catalogclient.DecodeModule(e.Value); err == nil {
fp2[m.Module] = mod.Fp2
}
if ok {
fp2[m.Module] = e.Fp
}
}
sort.Strings(placed)
@@ -1696,9 +1660,9 @@ func placedModules(s Store, node string) ([]string, map[string]string, error) {
}
// WatchedReads are the states the holder reads whole, each from the watched copy (ADR 0580 clause 2): the assignments,
// the wants, the indexer's modules, the haves and the published compositions, and the providers' entries, whose watch
// the wants, the haves and the published compositions, and the providers' entries, whose watch
// is also what rejudges a key.
var WatchedReads = append([]string{AssignmentsRead, WantsRead, ModulesRead, HaveRead, ComposedRead}, boundReads()...)
var WatchedReads = append([]string{AssignmentsRead, WantsRead, HaveRead, ComposedRead}, boundReads()...)
// OwnStates are the states the holder alone writes (ADR 0570 clause 1), read from the watched copy too (issue 1954: some
// 250 to 400 reads by key per composition): each of its own writes is taken into the copy at once (WatchedStates.Wrote),
@@ -1734,8 +1698,6 @@ func (w watchingStore) copied(state string) bool {
func (w watchingStore) Get(state, key string) (*Entry, error) {
get := w.ws.Get
switch {
case state == ManifestsRead:
get = w.ws.Once // a manifest by its digest never changes: read once, kept (ADR 0580 clause 1)
case !w.copied(state):
return w.Store.Get(state, key)
}
modules/mesh-provisioner/cmd/mesh-provisioner/holder_integration_test.go
+4 -22
View File
@@ -16,9 +16,6 @@ import (
"errors"
assignerclient "git.novox.be/novox/mesh-contracts/go/clients/meshmoduleassignerclient"
meshseat "git.novox.be/novox/mesh-contracts/go/seats/meshmoduleassigner"
catalogclient "git.novox.be/novox/mesh-delivery/contracts/go/clients/meshcatalogclient"
catalog "git.novox.be/novox/mesh-delivery/contracts/go/seats/meshcatalog"
builds "git.novox.be/novox/mesh-delivery/modules/mesh-delivery-builder/builds"
"strings"
"testing"
"time"
@@ -135,10 +132,7 @@ func TestTheHolderPlansProvidesAndAsksOnTheBus(t *testing.T) {
mem := goldenMesh(t, c)
published(t, mem, c)
vaultManifest := `{"module":"mesh-vault","version":"1","state":["bound"]}` // the vault keeps its outcomes as entries
vaultDigest := catalogclient.ManifestDigest([]byte(vaultManifest))
vaultValue, _ := catalogclient.EncodeManifest(catalog.ManifestsByDigestValue{Module: vaultModule, Manifest: vaultManifest})
mem.write(t, ManifestsRead, mustKey(catalogclient.ManifestsKey(vaultDigest)), string(vaultValue))
mem.write(t, ModulesRead, vaultModule, `{"v":1,"none2":"the test's","manifest":"`+vaultDigest+`"}`)
mem.cat.set(vaultModule, vaultManifest, "")
for state, keys := range mem.states {
for key, raw := range keys {
if err := s.Put(state, key, json.RawMessage(raw)); err != nil {
@@ -197,8 +191,8 @@ func TestTheHolderPlansProvidesAndAsksOnTheBus(t *testing.T) {
}
}
a := &asked{}
h := &Holder{Store: Watching(s, ws), Asks: a, Now: time.Now, Logf: t.Logf}
h.Compose = FromCompositions(h.Store)
h := &Holder{Store: Watching(s, ws), Catalog: mem.cat, Asks: a, Now: time.Now, Logf: t.Logf}
h.Compose = FromCompositions(h.Store, mem.cat)
state := func(key string) meshprovisionerclient.Provision {
e, _ := s.Get(ProvisionsState, key)
if e == nil {
@@ -230,21 +224,9 @@ func TestTheHolderPlansProvidesAndAsksOnTheBus(t *testing.T) {
settled(HaveRead, node+".postgres")
// The running build's manifest, as the builder keyed it: one that does not offer the provision keeps the key asked
// (ADR 0603 clause 5), one that does lets the have count.
buildsKey := mustKey(builds.BuilderBuildKey("postgres", fpsOf["postgres"]))
offered := func(provides string) {
raw := `{"module":"postgres","version":"1","provides":` + provides + `}`
digest := catalogclient.ManifestDigest([]byte(raw))
value, _ := catalogclient.EncodeManifest(catalog.ManifestsByDigestValue{Module: "postgres", Manifest: raw})
if err := s.Put(ManifestsRead, mustKey(catalogclient.ManifestsKey(digest)), json.RawMessage(value)); err != nil {
t.Fatal(err)
}
build, err := builds.EncodeBuilderBuild(builds.BuilderBuild{State: builds.BuildDone, Manifest: digest})
if err != nil {
t.Fatal(err)
}
if err := s.Put(BuildsRead, buildsKey, json.RawMessage(build)); err != nil {
t.Fatal(err)
}
mem.cat.setBuild("postgres", fpsOf["postgres"], raw)
}
offered(`["other"]`)
for i := 0; i < 2; i++ {
modules/mesh-provisioner/cmd/mesh-provisioner/holder_test.go
+6 -25
View File
@@ -6,9 +6,6 @@ import (
"fmt"
assignerclient "git.novox.be/novox/mesh-contracts/go/clients/meshmoduleassignerclient"
meshseat "git.novox.be/novox/mesh-contracts/go/seats/meshmoduleassigner"
catalogclient "git.novox.be/novox/mesh-delivery/contracts/go/clients/meshcatalogclient"
catalog "git.novox.be/novox/mesh-delivery/contracts/go/seats/meshcatalog"
builds "git.novox.be/novox/mesh-delivery/modules/mesh-delivery-builder/builds"
"go/ast"
"go/parser"
"go/token"
@@ -99,7 +96,7 @@ func newMesh() *mesh {
heldAt: map[string]map[string]string{}, unpublished: map[string][]string{},
heldNewest: map[string]map[string]string{}, unpublishedNewest: map[string][]string{}}
now := time.Date(2026, 10, 29, 22, 0, 0, 0, time.UTC)
m.h = &Holder{Store: m.s, Asks: m.r, Logf: func(string, ...any) {},
m.h = &Holder{Store: m.s, Catalog: m.s.cat, Asks: m.r, Logf: func(string, ...any) {},
Now: func() time.Time { now = now.Add(time.Second); return now },
Compose: func(_ context.Context, node string, keptAt map[string]string) Composition {
m.composed = append(m.composed, node)
@@ -112,11 +109,7 @@ func newMesh() *mesh {
for module, states := range map[string]string{"postgres": `["bound"]`, "keycloak": `["bound"]`, "mesh-vault": `["bound"]`,
"systemd-resolved": `[]`, "minio": `[]`} {
raw := `{"module":"` + module + `","version":"1","state":` + states + `}`
digest := catalogclient.ManifestDigest([]byte(raw))
value, _ := catalogclient.EncodeManifest(catalog.ManifestsByDigestValue{Module: module, Manifest: raw})
_ = m.s.Put(ManifestsRead, mustKey(catalogclient.ManifestsKey(digest)), json.RawMessage(value))
_ = m.s.Put(ModulesRead, mustKey(catalogclient.ModulesKey(module)),
json.RawMessage(`{"v":1,"none2":"a test","manifest":"`+digest+`","repository":"novox/mesh-catalog","commit":"c1"}`))
m.s.cat.set(module, raw, "")
}
return m
}
@@ -218,20 +211,12 @@ func TestAKeysStateFollowsItsProvidersOutcome(t *testing.T) {
// A have counts as the provider's outcome only from a build whose manifest offers the provision (ADR 0603 clause 5):
// refused when the build's manifest is known and does not offer it, provided when it does.
bfp := "src2:" + strings.Repeat("bb", 32)
buildKey := mustKey(builds.BuilderBuildKey("systemd-resolved", bfp))
for _, row := range []struct{ name, provides, state, why string }{
{"a build whose manifest does not offer it", `["other"]`, meshprovisionerclient.Asked, "offers split-dns"},
{"a build whose manifest offers it", `[{"name":"split-dns"}]`, meshprovisionerclient.Provided, ""},
} {
raw := `{"module":"systemd-resolved","version":"1","provides":` + row.provides + `}`
digest := catalogclient.ManifestDigest([]byte(raw))
value, _ := catalogclient.EncodeManifest(catalog.ManifestsByDigestValue{Module: "systemd-resolved", Manifest: raw})
_ = m.s.Put(ManifestsRead, mustKey(catalogclient.ManifestsKey(digest)), json.RawMessage(value))
build, err := builds.EncodeBuilderBuild(builds.BuilderBuild{State: builds.BuildDone, Manifest: digest})
if err != nil {
t.Fatal(err)
}
_ = m.s.Put(BuildsRead, buildKey, json.RawMessage(build))
m.s.cat.setBuild("systemd-resolved", bfp, raw)
m.compose(t, "g14")
if p := m.state(t, split); p.State != row.state || !strings.Contains(p.Why, row.why) {
t.Errorf("%s: %+v, want %s", row.name, p, row.state)
@@ -565,13 +550,9 @@ func TestTheHolderHoldsNoTimer(t *testing.T) {
}
}
// manifestOf puts a provider's manifest into the indexer's reads as the holder finds it.
// manifestOf puts a provider's manifest into the catalog half as the holder finds it.
func (m *mesh) manifestOf(module, raw string) {
digest := catalogclient.ManifestDigest([]byte(raw))
value, _ := catalogclient.EncodeManifest(catalog.ManifestsByDigestValue{Module: module, Manifest: raw})
_ = m.s.Put(ManifestsRead, mustKey(catalogclient.ManifestsKey(digest)), json.RawMessage(value))
_ = m.s.Put(ModulesRead, mustKey(catalogclient.ModulesKey(module)),
json.RawMessage(`{"v":1,"none2":"a test","manifest":"`+digest+`","repository":"novox/mesh-catalog","commit":"c1"}`))
m.s.cat.set(module, raw, "")
}
// A provider's outcome is read per offer: from its entries where the provider helper answers that offer, from its have
@@ -832,7 +813,7 @@ func TestARetirementIsJudgedOnlyFromAWholePicture(t *testing.T) {
s := goldenMesh(t, c)
published(t, s, c)
writeAssignments(t, s, c.Node, "store", "postgres", "mesh-provisioner-client", "restic", "de-spiegel")
got := FromCompositions(s)(context.Background(), c.Node, map[string]string{"photos": fp})
got := FromCompositions(s, s.cat)(context.Background(), c.Node, map[string]string{"photos": fp})
if !strings.Contains(got.AtWants.Held["de-spiegel"], "no build") || !slices.Equal(got.Unpublished, []string{"photos"}) {
t.Errorf("held %v, unpublished %v", got.AtWants.Held, got.Unpublished)
}
modules/mesh-provisioner/cmd/mesh-provisioner/main.go
+48 -15
View File
@@ -8,6 +8,7 @@ package main
import (
"context"
"encoding/json"
"fmt"
catalogclient "git.novox.be/novox/mesh-delivery/contracts/go/clients/meshcatalogclient"
"os"
@@ -76,6 +77,8 @@ func (runtimeAsks) Record(id string) ([]byte, bool, error) {
// loads is which of the reads a pass composes from have loaded once: their watch took every current value.
type loads struct {
// also says what else has not loaded, "" when nothing: the catalog half's first views.
also func() string
mu sync.Mutex
want []string
loaded map[string]bool
@@ -99,6 +102,11 @@ func (l *loads) missing() string {
out = append(out, s)
}
}
if l.also != nil {
if more := l.also(); more != "" {
out = append(out, more)
}
}
return strings.Join(out, ", ")
}
@@ -178,18 +186,23 @@ func slowest(took []Took) time.Duration {
const (
AssignmentsTool = assignerModule + ".assigner_read_assignments"
WantsTool = assignerModule + ".assigner_read_wants"
ModulesTool = indexerModule + ".indexer_read_modules"
ManifestsTool = indexerModule + ".indexer_read_manifests"
)
// newViews is the provisioner's pictures of the assigner's assignments and wants and the indexer's modules and manifests:
// The seat verbs the follower of the watched branches asks and the verb that reads an entry by its fingerprint (design
// 47 G §614), which the manifest invokes.
const (
RevisionVerb = catalogclient.RevisionVerb
RegistrationsVerb = "seat:mesh-catalog.registrations"
SourceVerb = catalogclient.SourceVerb
EntryVerb = catalogclient.EntryVerb
)
// newViews is the provisioner's pictures of the assigner's assignments and wants:
// each asked once of its owner's read tool, then followed by the owner's events (ADR 0615).
func newViews() *views.Reader {
v := &views.Reader{Module: Module, Logf: logf, Acts: meshhandactsclient.Recorder{Ask: stdio.Ask}}
v.Follow(assignerModule, "assignments", AssignmentsTool)
v.Follow(assignerModule, "wants", WantsTool)
v.Follow(indexerModule, "modules", ModulesTool)
v.Follow(indexerModule, "manifests", ManifestsTool)
return v
}
@@ -228,6 +241,15 @@ func newFindingsWriter() (*views.Writer[seat.Finding, seat.FindingCleared, seat.
func main() {
s := runtimeStore{}
v := newViews()
// Which modules exist, and each one's manifest, are the catalog half's, followed per watched branch by the seat's
// client on the one reader (design 47 G §196, §614).
branches, err := catalogclient.FollowBranches(v, stdio.Ask)
if err != nil {
logf("%v", err)
os.Exit(1)
}
branches.Logf = logf
cat := branchesHalf{branches}
findings, err := newFindingsWriter()
if err != nil {
logf("%v", err)
@@ -254,14 +276,31 @@ func main() {
os.Exit(1)
}
}()
v.Start(context.Background())
h := &Holder{Store: Watching(s, ws), Asks: runtimeAsks{}, Now: time.Now, Logf: logf}
h.Compose = FromCompositions(h.Store) // the node-engines' published compositions, from the copy (ADR 0593 §8)
h := &Holder{Store: Watching(s, ws), Catalog: cat, Asks: runtimeAsks{}, Now: time.Now, Logf: logf}
h.Compose = FromCompositions(h.Store, cat) // the node-engines' published compositions, from the copy (ADR 0593 §8)
w := &work{pending: map[string]bool{}, wake: make(chan struct{}, 1)}
// No pass judges before every read it composes from has loaded once (issue 940); each load wakes every node it
// held back, through its follower's handler.
loads := newLoads(AssignmentsRead, WantsRead, ComposedRead, HaveRead, ModulesRead)
loads := newLoads(AssignmentsRead, WantsRead, ComposedRead, HaveRead)
loads.also = func() string {
if err := cat.Ready(); err != nil {
return err.Error()
}
return ""
}
h.Loaded = loads.missing
// A module's entry changing in a branch's half wakes every node it touches, and the first view held lets the passes it
// held back run.
branches.Changed = func(_, module string, _ json.RawMessage, _ bool, _ time.Time) {
w.add(h.NodesOfModule(module)...)
w.add(loads.heldBack(h)...)
}
branches.Entries.Changed = func(row catalogclient.EntryRow) {
if row.Entry != nil {
w.add(h.NodesOfModule(row.Entry.Module)...)
}
}
v.Start(context.Background())
go w.run(h)
go starting(h)
// Each watch's handler runs after the copy took its change, so a composition it wakes reads the copy that change made.
@@ -276,12 +315,6 @@ func main() {
follow(WantsRead, func(c stdio.StateChange) error { w.add(NodesOfWant(c.Key)...); return nil })
follow(ComposedRead, func(c stdio.StateChange) error { w.add(NodesOfNodeKey(c.Key)...); return nil })
follow(HaveRead, func(c stdio.StateChange) error { w.add(NodesOfNodeKey(c.Key)...); return nil })
follow(ModulesRead, func(c stdio.StateChange) error {
if module, err := catalogclient.ParseModulesKey(c.Key); err == nil {
w.add(h.NodesOfModule(module)...)
}
return nil
})
for _, provider := range BoundReads {
// Into the copy the holder reads them from, its handler run after the copy took the change.
go ws.Follow(boundRead(provider), func(c stdio.StateChange) error {
modules/mesh-provisioner/cmd/mesh-provisioner/provisioner_test.go
+20 -8
View File
@@ -19,9 +19,13 @@ type memStore struct {
mu sync.Mutex
states map[string]map[string][]byte
puts int
// cat is the catalog half the holder reads, beside the states.
cat *fakeCatalog
}
func newStore() *memStore { return &memStore{states: map[string]map[string][]byte{}} }
func newStore() *memStore {
return &memStore{states: map[string]map[string][]byte{}, cat: newFakeCatalog()}
}
func (m *memStore) Get(state, key string) (*Entry, error) {
m.mu.Lock()
@@ -315,9 +319,17 @@ func TestTheManifestDeclaresWhatTheCodeKeeps(t *testing.T) {
if err := json.Unmarshal(raw, &m); err != nil {
t.Fatal(err)
}
// The builder's keys are read through its published package, taken by its series (ADR 0634 §1).
if m.Packages["git.novox.be/novox/mesh-delivery/modules/mesh-delivery-builder/builds"] != "0.1" {
t.Errorf("packages %v: want builds of series 0.1", m.Packages)
// Which modules exist and a build's manifest are the catalog half's (design 47 G §594): the manifest names neither the
// indexer's tools and views nor the builder's builds, and takes no package of the builder.
if len(m.Packages) != 0 {
t.Errorf("packages %v: the builder's keys are no longer read", m.Packages)
}
for _, list := range [][]string{m.Reads, m.Invokes, m.Consumes} {
for _, e := range list {
if strings.Contains(e, "indexer") || strings.Contains(e, "mesh-delivery-builder") {
t.Errorf("the manifest names %s, which only the catalog half answers now", e)
}
}
}
if m.Module != "mesh-provisioner" || m.RunsAs != "mesh-provisioner" || m.OwnSecrets["broker"] == "" {
t.Errorf("an account of its own: module %q, runs-as %q, own secrets %v", m.Module, m.RunsAs, m.OwnSecrets)
@@ -356,8 +368,8 @@ func TestTheManifestDeclaresWhatTheCodeKeeps(t *testing.T) {
// was made from) and makes no plan, so ADR 0593 §8 ("it composes nothing") holds.
// The assigner's assignments and wants and the indexer's modules and manifests are asked once of their owners' read
// tools and followed by their events, not read (ADR 0615 §1).
followed := []string{AssignmentsRead, WantsRead, ModulesRead, ManifestsRead}
for _, r := range append(slices.Clone(WatchedReads), BuildsRead) {
followed := []string{AssignmentsRead, WantsRead}
for _, r := range slices.Clone(WatchedReads) {
if slices.Contains(followed, r) {
if slices.Contains(m.Reads, r) {
t.Errorf("the manifest reads %s, which the holder follows by its owner's events", r)
@@ -368,9 +380,9 @@ func TestTheManifestDeclaresWhatTheCodeKeeps(t *testing.T) {
t.Errorf("the manifest does not read %s, which the holder reads", r)
}
}
if !slices.Equal(m.Invokes, []string{AssignmentsTool, WantsTool, ModulesTool, ManifestsTool, "seat:mesh-hand-acts.hand-act"}) ||
if !slices.Equal(m.Invokes, []string{AssignmentsTool, WantsTool, RevisionVerb, RegistrationsVerb, EntryVerb, SourceVerb, "seat:mesh-hand-acts.hand-act"}) ||
!slices.Equal(m.Consumes, []string{"mesh-delivery-assigner.views.assignments", "mesh-delivery-assigner.views.wants",
"mesh-delivery-indexer.views.modules", "mesh-delivery-indexer.views.manifests", "mesh-conditions.findings-asked"}) {
"mesh-catalog.catalog-revision", "mesh-catalog.registration-changed", "mesh-delivery.source-moved", "mesh-conditions.findings-asked"}) {
t.Errorf("invokes %v, consumes %v", m.Invokes, m.Consumes)
}
// It composes nothing (ADR 0593 §8): no config value is read to make a plan.
modules/mesh-provisioner/go.mod
-1
View File
@@ -14,7 +14,6 @@ require (
git.novox.be/novox/mesh-delivery/contracts/go/clients/meshcatalogkey v0.0.0-00010101000000-000000000000
git.novox.be/novox/mesh-delivery/contracts/go/seats/meshcatalog v0.0.0-00010101000000-000000000000
git.novox.be/novox/mesh-delivery/contracts/go/seats/meshwants v0.0.0-00010101000000-000000000000
git.novox.be/novox/mesh-delivery/modules/mesh-delivery-builder/builds v0.0.0-00010101000000-000000000000
git.novox.be/novox/mesh-sdk/go v0.0.0-00010101000000-000000000000
github.com/nats-io/nats.go v1.54.0
)
modules/mesh-provisioner/module.json
+7 -5
View File
@@ -19,7 +19,6 @@
],
"reads": [
"node-engine.have",
"mesh-delivery-builder.builds",
"mesh-host.composed",
"gitea.bound",
"keycloak.bound",
@@ -32,15 +31,18 @@
"invokes": [
"mesh-delivery-assigner.assigner_read_assignments",
"mesh-delivery-assigner.assigner_read_wants",
"mesh-delivery-indexer.indexer_read_modules",
"mesh-delivery-indexer.indexer_read_manifests",
"seat:mesh-catalog.revision",
"seat:mesh-catalog.registrations",
"seat:mesh-catalog.entry",
"seat:mesh-delivery.source",
"seat:mesh-hand-acts.hand-act"
],
"consumes": [
"mesh-delivery-assigner.views.assignments",
"mesh-delivery-assigner.views.wants",
"mesh-delivery-indexer.views.modules",
"mesh-delivery-indexer.views.manifests",
"mesh-catalog.catalog-revision",
"mesh-catalog.registration-changed",
"mesh-delivery.source-moved",
"mesh-conditions.findings-asked"
],
"tools": [
modules/mesh-self-check/cmd/mesh-self-check/catalogfeed.go
+53
View File
@@ -0,0 +1,53 @@
package main
// The self-check's reading of which modules exist (design 47 G §196, §594, §614): the catalog half, taken through
// meshcatalogclient's follower of the watched branches on the part's one views reader. A module's manifest is the one its
// catalog graph entry carries whole (§443), never the indexer's state `modules` or `manifests`.
import (
"fmt"
"strings"
"git.novox.be/novox/mesh-delivery/contracts/go/clients/meshcatalogclient"
catalog "git.novox.be/novox/mesh-delivery/contracts/go/seats/meshcatalog"
)
// CatalogHalf is every module's entry in the catalog half, as the self-check holds it.
type CatalogHalf interface {
// Modules is each module's entry; with a branch that has no first view yet it is an error naming it, never no module.
Modules() (map[string]catalog.CatalogEntry, error)
}
// branchesHalf is the CatalogHalf over the follower of the watched branches.
type branchesHalf struct{ b *meshcatalogclient.Branches }
func (h branchesHalf) Modules() (map[string]catalog.CatalogEntry, error) {
followed := h.b.Followed()
if len(followed) == 0 {
return nil, fmt.Errorf("the catalog half names no watched branch yet (the seat mesh-catalog's registrations are asked until they answer)")
}
out := map[string]catalog.CatalogEntry{}
var waiting []string
for _, branch := range followed {
c, ok := h.b.Catalog(branch)
if !ok {
continue
}
select {
case <-c.Ready():
default:
waiting = append(waiting, branch)
continue
}
half, _ := c.View() // a branch at none-yet is ready and empty (§614)
for m, e := range half.Entries {
if _, ok := out[m]; !ok { // the branch followed first names a module both hold
out[m] = e
}
}
}
if len(waiting) > 0 {
return nil, fmt.Errorf("the catalog half has no first view of %s yet", strings.Join(waiting, ", "))
}
return out, nil
}
modules/mesh-self-check/cmd/mesh-self-check/catalogfeed_test.go
+98
View File
@@ -0,0 +1,98 @@
package main
import (
"context"
"encoding/json"
"fmt"
"strings"
"sync"
"testing"
"time"
"git.novox.be/novox/mesh-delivery/contracts/go/clients/meshcatalogclient"
catalog "git.novox.be/novox/mesh-delivery/contracts/go/seats/meshcatalog"
delivery "git.novox.be/novox/mesh-delivery/contracts/go/seats/meshdelivery"
"git.novox.be/novox/mesh-sdk/go/views"
)
// wiringOf is a part's reader in miniature: it keeps the followers adopted, so a test starts them as the reader's Start does.
type wiringOf struct {
mu sync.Mutex
fs []*views.Follower
}
func (w *wiringOf) Adopt(f *views.Follower, _, _ string, _ func(context.Context, []byte) error) error {
w.mu.Lock()
defer w.mu.Unlock()
w.fs = append(w.fs, f)
return nil
}
func (w *wiringOf) Drop(string, string) error { return nil }
// Design 47 G §196, §594: which modules exist, and the manifest of each, are the catalog half's; a branch with no first
// view yet is an error naming it, never no module, and an entry's manifest is the one it carries whole.
func TestCatalogHalfIsTheWatchedBranchesNewestRevision(t *testing.T) {
ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
defer cancel()
answered := false
var mu sync.Mutex
call := func(key string, _ any) (json.RawMessage, error) {
mu.Lock()
defer mu.Unlock()
switch key {
case meshcatalogclient.RegistrationsVerb:
return json.Marshal(catalog.RegistrationsAnswer{Registrations: []catalog.Registration{{Module: "m", Version: 1, Branches: []string{"main"}}}})
case meshcatalogclient.SourceVerb:
return json.Marshal(delivery.SourceAnswer{})
case meshcatalogclient.RevisionVerb:
if !answered {
return nil, fmt.Errorf("the holder is away")
}
return json.Marshal(catalog.RevisionAnswer{Revision: &catalog.CatalogRevision{Revision: 1, Kind: meshcatalogclient.KindTrunk},
Entries: []catalog.CatalogEntry{{Module: "pg", Manifest: map[string]any{"module": "pg"}}}})
}
return nil, fmt.Errorf("asked %s", key)
}
w := &wiringOf{}
b, err := meshcatalogclient.NewBranches(w, call)
if err != nil {
t.Fatal(err)
}
h := branchesHalf{b}
if _, err := h.Modules(); err == nil || !strings.Contains(err.Error(), "no watched branch") {
t.Fatalf("before the registrations: %v, want an error naming the missing branch", err)
}
go func() { _ = w.fs[0].Start(ctx) }() // the registrations
for len(b.Followed()) == 0 {
select {
case <-ctx.Done():
t.Fatal("the registrations' branch is never followed")
case <-time.After(10 * time.Millisecond):
}
}
if _, err := h.Modules(); err == nil || !strings.Contains(err.Error(), "no first view of main") {
t.Fatalf("with no first view: %v, want an error naming the branch, never no module", err)
}
mu.Lock()
answered = true
mu.Unlock()
c, _ := b.Catalog("main")
if err := c.Start(ctx); err != nil {
t.Fatal(err)
}
got, err := h.Modules()
if err != nil || len(got) != 1 || got["pg"].Manifest["module"] != "pg" {
t.Fatalf("with the first view: %v, %v", got, err)
}
}
// D13 reads each manifest from the catalog entry that carries it, and a catalog half without a view is an error, never a
// mesh with no module declaring data.
func TestD13ReadsTheCatalogHalf(t *testing.T) {
mesh := dataMesh(func(*dataRig) {})
mesh.modulesErr = fmt.Errorf("the catalog half has no first view of main yet")
c := &Checker{Store: mesh, Catalog: mesh, Now: func() time.Time { return dataNow }}
if _, err := probeData(context.Background(), c); err == nil || !strings.Contains(err.Error(), "no first view") {
t.Fatalf("D13 with no catalog view: %v", err)
}
}
modules/mesh-self-check/cmd/mesh-self-check/main.go
+11 -3
View File
@@ -18,6 +18,7 @@ import (
"git.novox.be/novox/mesh-contracts/health-and-repair/go/clients/meshhandactsclient"
meshconditions "git.novox.be/novox/mesh-contracts/health-and-repair/go/seats/meshconditions"
"git.novox.be/novox/mesh-delivery/contracts/go/clients/meshcatalogclient"
"git.novox.be/novox/mesh-sdk/go/views"
)
@@ -76,7 +77,16 @@ func NewChecker() *Checker {
os.Exit(1)
}
c.Writer = w
c.Store = runtimeStore{views: newViews(w)}
v := newViews(w)
// Which modules exist is the catalog half's, followed per watched branch by the seat's client (design 47 G §196, §614).
b, err := meshcatalogclient.FollowBranches(v, func(key string, body any) (json.RawMessage, error) { return stdio.Ask(key, body) })
if err != nil {
logf("%v", err)
os.Exit(1)
}
b.Logf = logf
c.Store = runtimeStore{views: v}
c.Catalog = branchesHalf{b}
return c
}
@@ -87,8 +97,6 @@ func newViews(w *FindingsWriter) *views.Reader {
r := &views.Reader{Module: Module, Logf: logf, Findings: w, Acts: meshhandactsclient.Recorder{Ask: stdio.Ask}}
r.Follow(assignerModule, placing.StateAssignments, AssignmentsTool)
r.Follow(assignerModule, placing.StateSeatHolders, SeatHoldersTool)
r.Follow(indexerModule, "modules", ModulesTool)
r.Follow(indexerModule, "manifests", ManifestsTool)
return r
}
modules/mesh-self-check/cmd/mesh-self-check/probes.go
+12 -5
View File
@@ -34,12 +34,13 @@ const (
// Invokes are the tools and seat verbs the probes ask, and Reads the states they read: the manifest's `invokes` and
// `reads`, which its test holds to these.
var (
Invokes = []string{StreamsTool, ClosedTool, SignalsVerb, BannedVerb, AssignmentsTool, SeatHoldersTool, ModulesTool, ManifestsTool, HandActVerb}
Reads = []string{PresenceRead, HealthRead, ReadingsRead, GoodRead}
Invokes = []string{StreamsTool, ClosedTool, SignalsVerb, BannedVerb, AssignmentsTool, SeatHoldersTool,
RevisionVerb, RegistrationsVerb, SourceVerb, HandActVerb}
Reads = []string{PresenceRead, HealthRead, ReadingsRead, GoodRead}
// Consumes are the views families the probes follow (ADR 0615): the indexer's and the assigner's states are asked
// once, by their owners' read tools, and then followed by their events, never read.
Consumes = []string{"mesh-delivery-assigner.views.assignments", "mesh-delivery-assigner.views.seat-holders",
"mesh-delivery-indexer.views.modules", "mesh-delivery-indexer.views.manifests"}
"mesh-catalog.catalog-revision", "mesh-catalog.registration-changed", "mesh-delivery.source-moved"}
)
// HandActVerb records a re-read of a first view by hand (ADR 0615 §6).
@@ -49,8 +50,14 @@ const HandActVerb = "seat:mesh-hand-acts.hand-act"
const (
AssignmentsTool = "mesh-delivery-assigner.assigner_read_assignments"
SeatHoldersTool = "mesh-delivery-assigner.assigner_read_seat_holders"
ModulesTool = "mesh-delivery-indexer.indexer_read_modules"
ManifestsTool = "mesh-delivery-indexer.indexer_read_manifests"
)
// The seat verbs the follower of the watched branches asks (design 47 G §614): the catalog half's first view, the
// registrations' watched branches and, before the first registrations, the branches the gate's source answers.
const (
RevisionVerb = "seat:mesh-catalog.revision"
RegistrationsVerb = "seat:mesh-catalog.registrations"
SourceVerb = "seat:mesh-delivery.source"
)
// Registry is the self-check's registry, in to-be 45's order. **The registry is the design's live form**: a probe
modules/mesh-self-check/cmd/mesh-self-check/probes_data.go
+13 -23
View File
@@ -4,7 +4,7 @@ package main
// declares is measured, is there, holds what it held, and is backed up within its bound. The controller composed what
// every machine declares and asked each machine's backup holder what it measured; the facts now live on the bus: the
// readings disk-load keeps (its state `readings`, with the item's seven-day peak), the last good night per item restic
// keeps (its state `good`), and the manifests the indexer holds (for the bounds an item declares). Each kind, key and
// keeps (its state `good`), and the manifests the catalog half's entries carry (for the bounds an item declares). Each kind, key and
// word is the controller's (mesh-controller cmd/mesh-controller/data.go), so the condition the keeper raises from it is
// the one the controller raised.
@@ -27,10 +27,8 @@ import (
// The states D13 reads (declared under `reads`).
const (
ReadingsRead = diskreadings.ReadingsRead
GoodRead = backupgood.GoodRead
ModulesRead = indexerModule + ".modules"
ManifestsRead = indexerModule + ".manifests"
ReadingsRead = diskreadings.ReadingsRead
GoodRead = backupgood.GoodRead
)
// The kinds D13 raises.
@@ -63,7 +61,7 @@ type dataFacts struct {
// keptStates are the states D13 judges from memory (issue 1952): read one key at a time through the runtime, each
// read reopening its bucket, a run of D13 never finished within its bound. Each is watched once and kept as it changes.
var keptStates = []string{ReadingsRead, GoodRead, ModulesRead, ManifestsRead}
var keptStates = []string{ReadingsRead, GoodRead}
// view is one state kept in memory from its watch: ready once the watch handed over every current value, err when it
// was refused (then the next read asks again).
@@ -152,26 +150,18 @@ func (c *Checker) dataFacts(ctx context.Context) (dataFacts, error) {
f.good[backupgood.GoodKey(g.Node, g.Module, g.Item)] = g.LastGood
}
}
for name, raw := range states[ModulesRead] {
// Only the manifest's digest of the indexer's module, and the manifest's text: what the bounds are read from.
var m struct {
Manifest string `json:"manifest"`
}
if json.Unmarshal(raw, &m) != nil || m.Manifest == "" {
continue
}
mraw, ok := states[ManifestsRead][m.Manifest]
if !ok {
continue
}
var doc struct {
JSON string `json:"manifest"`
}
if json.Unmarshal(mraw, &doc) != nil {
// Each module's manifest is the one its catalog graph entry carries whole (design 47 G §594).
modules, err := c.Catalog.Modules()
if err != nil {
return f, err
}
for name, e := range modules {
raw, err := json.Marshal(e.Manifest)
if err != nil || len(e.Manifest) == 0 {
continue
}
var man catalogue.Manifest
if json.Unmarshal([]byte(doc.JSON), &man) == nil {
if json.Unmarshal(raw, &man) == nil {
f.manifests[name] = man
}
}
modules/mesh-self-check/cmd/mesh-self-check/probes_test.go
+9 -7
View File
@@ -15,7 +15,6 @@ import (
nodeseat "git.novox.be/novox/mesh-contracts/go/seats/nodeengine"
"git.novox.be/novox/mesh-contracts/health-and-repair/go/clients/meshconditionsclient"
watchdogsclient "git.novox.be/novox/mesh-contracts/health-and-repair/go/clients/meshwatchdogsclient"
catalogclient "git.novox.be/novox/mesh-delivery/contracts/go/clients/meshcatalogclient"
catalog "git.novox.be/novox/mesh-delivery/contracts/go/seats/meshcatalog"
"git.novox.be/novox/mesh-catalog/modules/disk-load/diskreadings"
@@ -214,7 +213,7 @@ func TestTheBusProbes(t *testing.T) {
agents(map[string]string{"novox": "old"})},
}
for _, tc := range cases {
c := &Checker{Ask: tc.bus.ask, Store: tc.mesh, Now: func() time.Time { return dataNow }}
c := &Checker{Ask: tc.bus.ask, Store: tc.mesh, Catalog: tc.mesh, Now: func() time.Time { return dataNow }}
if tc.mesh == nil {
c.Store = newStore()
}
@@ -222,7 +221,7 @@ func TestTheBusProbes(t *testing.T) {
// D13 judges the states it reads from memory, kept by watching them, never one key at a time through the runtime,
// which did not finish within its bound (issue 1952).
if strings.HasPrefix(tc.name, "D13") {
for _, s := range []string{ReadingsRead, GoodRead, ModulesRead, ManifestsRead} {
for _, s := range []string{ReadingsRead, GoodRead} {
if n := tc.mesh.gets[s]; n > 0 {
t.Errorf("%s: D13 read %s one key at a time %d times", tc.name, s, n)
}
@@ -244,7 +243,7 @@ func TestTheBusProbes(t *testing.T) {
}
// What changes after the watch began is judged at the next run, from memory: every reading gone is seen.
mesh := dataMesh(func(*dataRig) {})
c := &Checker{Store: mesh, Now: func() time.Time { return dataNow }}
c := &Checker{Store: mesh, Catalog: mesh, Now: func() time.Time { return dataNow }}
if found, err := probeData(context.Background(), c); err != nil || len(found) != 0 {
t.Fatalf("D13 on a held mesh: %v, %v", keysOf(found), err)
}
@@ -452,9 +451,12 @@ func dataMesh(change func(*dataRig)) *memStore {
man.Data.Own = append(man.Data.Own, it)
}
raw, _ := json.Marshal(man)
digest := catalogclient.ManifestDigest(raw)
_ = r.s.Put(ManifestsRead, digest, catalog.ManifestsByDigestValue{V: 1, Module: module, Manifest: string(raw)})
_ = r.s.Put(ModulesRead, module, catalog.ModulesByModuleValue{V: 1, Manifest: digest})
var whole map[string]any
_ = json.Unmarshal(raw, &whole)
if r.s.modules == nil {
r.s.modules = map[string]catalog.CatalogEntry{}
}
r.s.modules[module] = catalog.CatalogEntry{Module: module, Manifest: whole}
}
declare("pg", map[string]catalogue.DataItem{
"data": {Class: catalogue.ClassValuable, Within: "48h", Active: "24h"},
modules/mesh-self-check/cmd/mesh-self-check/selfcheck.go
+2
View File
@@ -110,6 +110,8 @@ type (
// Checker is the self-check: the registry and what its probes reach.
type Checker struct {
Store Store
// Catalog is the catalog half: which modules exist and the manifest each entry carries (design 47 G §594).
Catalog CatalogHalf
// Enabled is the setting `probes`: the ids of the probes that run. Nil runs every probe in the registry; one not
// listed runs nothing and raises nothing (the operator, 2026-10-30).
Enabled func() []string
modules/mesh-self-check/cmd/mesh-self-check/selfcheck_test.go
+7
View File
@@ -13,6 +13,7 @@ import (
"time"
"git.novox.be/novox/mesh-contracts/health-and-repair/go/clients/meshconditionsclient"
catalog "git.novox.be/novox/mesh-delivery/contracts/go/seats/meshcatalog"
stdio "git.novox.be/novox/mesh-sdk/go"
"git.novox.be/novox/mesh-sdk/go/views"
)
@@ -27,8 +28,14 @@ type memStore struct {
watches map[string][]func(key string, value []byte, gone bool)
keysErr error
events []string // the subjects the finding writer said its events on
// modules are the catalog half's entries (the CatalogHalf this store answers as the tests' own).
modules map[string]catalog.CatalogEntry
modulesErr error
}
// Modules is the catalog half, as the tests give it.
func (s *memStore) Modules() (map[string]catalog.CatalogEntry, error) { return s.modules, s.modulesErr }
func newStore() *memStore {
return &memStore{kv: map[string]map[string][]byte{}, puts: map[string]int{}, gets: map[string]int{},
watches: map[string][]func(string, []byte, bool){}}
modules/mesh-self-check/module.json
+6 -4
View File
@@ -23,8 +23,9 @@
"seat:node-intrusion-prevention.banned",
"mesh-delivery-assigner.assigner_read_assignments",
"mesh-delivery-assigner.assigner_read_seat_holders",
"mesh-delivery-indexer.indexer_read_modules",
"mesh-delivery-indexer.indexer_read_manifests",
"seat:mesh-catalog.revision",
"seat:mesh-catalog.registrations",
"seat:mesh-delivery.source",
"seat:mesh-hand-acts.hand-act"
],
"reads": [
@@ -36,8 +37,9 @@
"consumes": [
"mesh-delivery-assigner.views.assignments",
"mesh-delivery-assigner.views.seat-holders",
"mesh-delivery-indexer.views.modules",
"mesh-delivery-indexer.views.manifests",
"mesh-catalog.catalog-revision",
"mesh-catalog.registration-changed",
"mesh-delivery.source-moved",
"mesh-conditions.findings-asked"
],
"state": [