W16.3: mesh-provisioner reads modules and manifests from the catalog half through meshcatalogclient's branch follower (milestone 47/G)

Builds: design 47 G §196, §594, §614; lead answers A-5, A-15

Claude-Session: https://claude.ai/code/session_01NcbgfGKaxw984XyAPfLm4h
This commit is contained in:
2026-11-11 23:24:35 +01:00
parent 095e23ecda
commit 0098960b5a
11 changed files with 359 additions and 166 deletions
@@ -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
}
@@ -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)
}
@@ -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)
}
}
@@ -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)
@@ -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)
}
@@ -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++ {
@@ -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)
}
@@ -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 {
@@ -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.
-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
)
+7 -8
View File
@@ -1,9 +1,6 @@
{
"module": "mesh-provisioner",
"version": "2",
"packages": {
"git.novox.be/novox/mesh-delivery/modules/mesh-delivery-builder/builds": "0.1"
},
"runs-as": "mesh-provisioner",
"claims": [
{
@@ -19,7 +16,6 @@
],
"reads": [
"node-engine.have",
"mesh-delivery-builder.builds",
"mesh-host.composed",
"gitea.bound",
"keycloak.bound",
@@ -32,15 +28,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": [