issue 1840/S3: keep each node's declared packages on the bus, so the pacman verbs can say what a node declares

This commit is contained in:
2026-10-22 14:23:58 +02:00
parent cc8a74fce4
commit 485eb44de4
13 changed files with 360 additions and 9 deletions
@@ -0,0 +1,61 @@
//go:build integration
package main
import (
"encoding/json"
"testing"
"git.novox.be/novox/mesh-sdk/go/catalogue"
"github.com/novox/mesh-controller/internal/inventory"
)
// 1840/AC5 (ADR 0507 §5): the source of the declared-packages mirror composes each node and names, for a package two modules
// declare and one declared absent, the modules that declare it; the same records give the same bytes on a second pass, so
// the mirror writes nothing; a node with no package has a key that says so.
func TestTheDeclaredPackagesMirrorNamesWhatEachNodesModulesDeclare(t *testing.T) {
open := aMesh(t)
ctx := t.Context()
register(t, open, catalogue.Manifest{Module: "viewer", Version: "1", Resources: []map[string]any{
{"id": "pkg-curl", "type": "package", "package": "curl"},
{"id": "pkg-nano", "type": "package", "package": "nano", "absent": true}}})
register(t, open, catalogue.Manifest{Module: "fetcher", Version: "1", Resources: []map[string]any{
{"id": "pkg-curl", "type": "package", "package": "curl"}}})
for _, module := range []string{"viewer", "fetcher"} {
if _, err := assign(ctx, open, "anchor", module); err != nil {
t.Fatalf("assign anchor %s: %v", module, err)
}
}
source := declaredPackagesMirror(open)
first, err := source(ctx)
if err != nil {
t.Fatal(err)
}
var got inventory.DeclaredPackagesMirrored
if err := json.Unmarshal(first["anchor"], &got); err != nil {
t.Fatalf("anchor has no value: %v", err)
}
want := []inventory.PackageDeclared{{Package: "curl", Modules: []string{"fetcher", "viewer"}},
{Package: "nano", Modules: []string{"viewer"}, Absent: true}}
if len(got.Packages) != len(want) {
t.Fatalf("anchor declares %+v, want %+v", got.Packages, want)
}
for i, p := range want {
if g := got.Packages[i]; g.Package != p.Package || g.Absent != p.Absent || len(g.Modules) != len(p.Modules) ||
(len(p.Modules) > 0 && g.Modules[0] != p.Modules[0]) {
t.Errorf("package %d is %+v, want %+v", i, g, p)
}
}
if string(first["laptop"]) != `{"node":"laptop","packages":[]}` {
t.Errorf("a node declaring none is %s", first["laptop"])
}
second, err := source(ctx)
if err != nil {
t.Fatal(err)
}
for node, value := range first {
if string(second[node]) != string(value) {
t.Errorf("the value of %s changed between two passes over the same records:\n%s\n%s", node, value, second[node])
}
}
}
+1 -1
View File
@@ -281,7 +281,7 @@ func serve(ctx context.Context) (err error) {
// The provisioner's plan of each handed module, as its machine was last sent it (design 57 P6), in the module's
// assignment.
inv.SetProvisioningPlans(inv.SentProvisioningPlans)
go mirroringRecords(ctx, inv, bus.Conn, bus)
go mirroringRecords(ctx, open, inv, bus.Conn, bus)
// And the bucket of the files that travel by digest, judged against what each machine's latest declaration names (ADR 0486 §3).
go sweepingDeclarationFiles(ctx, inv, bus.Conn)
// And the key of each plan in its build phase, which holds the build seat's checks while a delivery builds (ADR 0483 clause 2).
+38 -3
View File
@@ -7,6 +7,7 @@ import (
"sort"
"time"
"git.novox.be/novox/mesh-sdk/go/compose"
"github.com/nats-io/nats.go"
"github.com/nats-io/nats.go/jetstream"
@@ -29,7 +30,7 @@ import (
const mirrorEvery = 10 * time.Second
// mirroringRecords keeps the mirrors current for as long as this controller serves: ctx ends when it stops.
func mirroringRecords(ctx context.Context, inv *inventory.Inventory, conn *nats.Conn, announce recordmirror.Announcer) {
func mirroringRecords(ctx context.Context, open *stores, inv *inventory.Inventory, conn *nats.Conn, announce recordmirror.Announcer) {
api, err := jetstream.New(conn)
if err != nil {
fmt.Printf("the record mirrors are not kept: %v\n", err)
@@ -44,8 +45,9 @@ func mirroringRecords(ctx context.Context, inv *inventory.Inventory, conn *nats.
return e
}
sources := map[string]func(context.Context) (map[string][]byte, error){
recordmirror.DeclaredConfig.Name: inv.DeclaredConfigMirror,
recordmirror.Predefined.Name: inv.PredefinedMirror,
recordmirror.DeclaredConfig.Name: inv.DeclaredConfigMirror,
recordmirror.Predefined.Name: inv.PredefinedMirror,
recordmirror.DeclaredPackages.Name: declaredPackagesMirror(open),
}
writers := map[string]*recordmirror.Writer{}
failing := map[string]string{}
@@ -176,3 +178,36 @@ func keepingAssignments(ctx context.Context, inv *inventory.Inventory, announce
}
events.known = now
}
// declaredPackagesMirror is the source of the declared-packages mirror (ADR 0507 §5): each node's composition, asked as a
// question that makes nothing, and the packages it declares with the modules that declare them. A node whose own set does
// not compose has no key, and any other failure to read is the pass's error, so a key is never purged for a store that
// could not be read.
func declaredPackagesMirror(open *stores) func(context.Context) (map[string][]byte, error) {
return func(ctx context.Context) (map[string][]byte, error) {
in, err := inputFor(ctx, open)
if err != nil {
return nil, err
}
// Which module owns each resource is carried only for a node-engine that reads it, so every node is asked as one that does.
asked := in.Input
asked.Nodes = append([]compose.Node(nil), in.Nodes...)
for i := range asked.Nodes {
asked.Nodes[i].ReadsModules = true
}
out := map[string][]byte{}
for _, n := range in.sortedNodes() {
d, _, err := compose.ComposeIn(asked, n.Name, compose.Foreseeing)
if compose.IsUnresolvable(err) {
continue
}
if err != nil {
return nil, fmt.Errorf("the packages %s declares cannot be read: %w", n.Name, err)
}
if out[n.Name], err = inventory.DeclaredPackagesValue(n.Name, d.Resources, d.ModuleOf); err != nil {
return nil, err
}
}
return out, nil
}
}