Merge branch 'b6/registry' into feat/registry-milestone

cmd/mesh-host/nodeagent.go

@@ -593,10 +593,12 @@ func nodeAgent(ctx context.Context, opts options, signer ed25519.PublicKey, trus
 		// composer hands nothing new meanwhile, and the node says it has waited five minutes (novox/hq ADR 0636 §1).
 		seats, err := startSeatList(run, session, node, say)
 		if err != nil {
-			say(err.Error())
+			// No reader is no gate that lets everything through: the node acts on nothing new, as one that waits does.
+			say(err.Error() + "; the node composes nothing new until it is started again and reads the list (novox/hq ADR 0636 §1)")
 		}
 		hooks := composerHooks{
-			seatList: seats,
+			seatList:       seats,
+			seatListFailed: err != nil,
 			running: func() []compositions.Composed {
 				if now := own.get(); now != nil {
 					return runsAsWritten(now.Everything(), opts.state)
@@ -639,6 +641,7 @@ func nodeAgent(ctx context.Context, opts options, signer ed25519.PublicKey, trus
 		// its own key, `<node>.node`, under candidates.
 		shadowHooks := hooks
 		shadowHooks.refused, shadowHooks.kept, shadowHooks.certify, shadowHooks.seatList = nil, nil, nil, nil
+		shadowHooks.seatListFailed = false
 		shadowEnv := composerEnv(session, node, say, shadowHooks)
 		shadow := selfcompose.NewShadow(shadowEnv,
 			func(ctx context.Context, answers map[string]nodeengine.CandidateAnswer) {
@@ -994,6 +997,8 @@ type composerHooks struct {
 	certify        func(ctx context.Context) error
 	// seatList is this run's reader of the seat list: the composer hands nothing new until it was read (ADR 0636 §1).
 	seatList *seatListState
+	// seatListFailed says the reader could not start: the composer then holds like a reader that waits (selfcompose.Unread).
+	seatListFailed bool
 }
 
 // selfComposer is the node's own composer: it composes what the seat's placement places on the node, and nothing while
@@ -1047,6 +1052,8 @@ func composerEnv(session *link.Session, node string, say func(string), hooks com
 		VaultAuthority: hooks.vaultAuthority, Certify: hooks.certify}
 	if hooks.seatList != nil {
 		env.SeatList, env.SeatListChanged = hooks.seatList.reader, hooks.seatList.changed
+	} else if hooks.seatListFailed {
+		env.SeatList = selfcompose.Unread{}
 	}
 	return env
 }

cmd/mesh-host/seatlist_unread_test.go

@@ -0,0 +1,22 @@
+package main
+
+import (
+	"testing"
+
+	"github.com/novox/mesh-host/internal/selfcompose"
+)
+
+// ADR 0636 §1: when the node's reader of the seat list could not start, the composer is gated by a reader that never
+// allows, where a missing reader would let the node act on everything; the shadow, which applies nothing, stays ungated.
+func TestAReaderThatCouldNotStartHoldsTheComposer(t *testing.T) {
+	failed := composerEnv(nil, "n1", func(string) {}, composerHooks{seatListFailed: true})
+	if failed.SeatList == nil || failed.SeatList.MayAct() || failed.SeatList.MayAct("node-engine") {
+		t.Fatalf("the composer of a node whose reader did not start is gated by %v, want a gate that allows nothing", failed.SeatList)
+	}
+	if _, ok := failed.SeatList.(selfcompose.Unread); !ok {
+		t.Errorf("gate is %T", failed.SeatList)
+	}
+	if got := composerEnv(nil, "n1", func(string) {}, composerHooks{}).SeatList; got != nil {
+		t.Errorf("a composer with no reader asked for gates by %v, want none (the shadow)", got)
+	}
+}

internal/selfcompose/selfcompose.go

@@ -28,6 +28,7 @@ import (
 	"git.novox.be/novox/mesh-delivery/contracts/go/clients/meshcatalogclient"
 	"git.novox.be/novox/mesh-delivery/contracts/go/seats/meshcatalog"
 	"git.novox.be/novox/mesh-delivery/modules/mesh-delivery-builder/builds"
+	"git.novox.be/novox/mesh-sdk/go/declaration"
 	"github.com/novox/mesh-host/internal/buckets"
 	"github.com/novox/mesh-host/internal/compositions"
 	engcomp "github.com/novox/mesh-host/internal/compositions"
@@ -1028,8 +1029,11 @@ func (c *Composer) compose(ctx context.Context) []agent.Update {
 		out = append(out, agent.Update{Key: key, Value: raw, Revision: c.rev})
 	}
 	for key := range c.handed {
-		if _, module, _, _, err := engcomp.ParseKey(key); err == nil && module != "" {
-			if _, isHeld := held[module]; isHeld {
+		if _, module, _, isNode, err := engcomp.ParseKey(key); err == nil {
+			if isNode {
+				module = agent.NodeModule // the node's own composition is held under its scope
+			}
+			if _, isHeld := held[module]; isHeld && module != "" {
 				continue // held: what was handed of it stays, so the node keeps running it
 			}
 		}
@@ -1784,6 +1788,18 @@ const WaitsForTheSeatList = "the node has not read the seat list, so it composes
 // cannot judge before the list was read, so only the last applied one runs (ADR 0636 §1).
 const WaitsForTheSeatListOwn = "the node has not read the seat list, so it composes nothing new of itself and keeps what runs (novox/hq ADR 0636 §1)"
 
+// WaitsForTheSeatListOwnPart is why the rest of the node's own composition is held: of it the node composes what the holder
+// of the seat mesh-catalog on this node needs, so that holder can start and the list can be read (ADR 0636 §1).
+const WaitsForTheSeatListOwnPart = "the node has not read the seat list, so it composes of itself only what the holder of the seat mesh-catalog needs " +
+	"and keeps what runs of the rest (novox/hq ADR 0636 §1)"
+
+// Unread is the gate of a node whose reader of the seat list could not start: it has no list and can never read one, so it
+// acts on nothing new, as a reader that waits does, where a nil gate would act on everything (ADR 0636 §1).
+type Unread struct{}
+
+// MayAct is false for every seat.
+func (Unread) MayAct(...string) bool { return false }
+
 // holdUntilListed takes out of compositions every module's composition that is new (no running composition has its
 // digest) while the reader may not act on its seats, and names the module in held: it keeps running what it runs, and
 // the pass after the list was read hands it. The node's own composition is held the same way, whatever it folds, under
@@ -1808,6 +1824,16 @@ func (c *Composer) holdUntilListed(ctx context.Context, r reader, compositions m
 				continue
 			}
 			module, why = agent.NodeModule, WaitsForTheSeatListOwn
+			// The list is read from the holder of the seat mesh-catalog, which the node's own composition may have to
+			// start: its part is composed whatever the list says, over what the node's own composition runs, and only that.
+			if part, ok := c.holderPart(ctx, r, comp, running); ok {
+				compositions[key] = part
+				if held == nil {
+					held = map[string]string{}
+				}
+				held[module] = WaitsForTheSeatListOwnPart
+				continue
+			}
 		} else if c.env.SeatList.MayAct(c.seatsOf(ctx, r, module)...) {
 			continue
 		}
@@ -1822,6 +1848,105 @@ func (c *Composer) holdUntilListed(ctx context.Context, r reader, compositions m
 	return held
 }
 
+// holderPart is the node's own composition as it runs with, of the new one comp, what the holders of the seat mesh-catalog
+// on this node need: their resources and their accounts, each replacing the one of the same id or name or following the
+// rest. False when the new composition has nothing of a holder: the caller holds the composition whole then.
+func (c *Composer) holderPart(ctx context.Context, r reader, comp compositions.Composed, running []compositions.Composed) (compositions.Composed, bool) {
+	holders := c.claimants(ctx, r, meshcatalog.Seat)
+	if len(holders) == 0 {
+		return compositions.Composed{}, false
+	}
+	out := compositions.Composed{V: comp.V, Node: comp.Node, Plan: comp.Plan}
+	for _, run := range running {
+		if run.Module == "" {
+			out = run
+			out.Resources = slices.Clone(run.Resources)
+			out.Filled = slices.Clone(run.Filled)
+			out.Own.Identities = slices.Clone(run.Own.Identities)
+			break
+		}
+	}
+	belongs := func(module, id string) bool {
+		for _, h := range holders {
+			if module == h || strings.HasPrefix(id, h+".") {
+				return true
+			}
+		}
+		return false
+	}
+	touched := false
+	paths := map[string]bool{}
+	for _, raw := range comp.Resources {
+		var head struct{ ID, Module, Type, Path string }
+		if json.Unmarshal(raw, &head) != nil || !belongs(head.Module, head.ID) {
+			continue
+		}
+		touched = true
+		if head.Type == "file" {
+			paths[head.Path] = true
+		}
+		at := slices.IndexFunc(out.Resources, func(have json.RawMessage) bool {
+			var h struct{ ID string }
+			return json.Unmarshal(have, &h) == nil && h.ID == head.ID
+		})
+		if at >= 0 {
+			out.Resources[at] = raw
+		} else {
+			out.Resources = append(out.Resources, raw)
+		}
+	}
+	for _, id := range comp.Own.Identities {
+		if !belongs(id.Module, "") {
+			continue
+		}
+		touched = true
+		at := slices.IndexFunc(out.Own.Identities, func(have declaration.Identity) bool { return have.Module == id.Module && have.Name == id.Name })
+		if at >= 0 {
+			out.Own.Identities[at] = id
+		} else {
+			out.Own.Identities = append(out.Own.Identities, id)
+		}
+	}
+	for _, p := range comp.Filled {
+		if paths[p] && !slices.Contains(out.Filled, p) {
+			out.Filled = append(out.Filled, p)
+		}
+	}
+	return out, touched
+}
+
+// claimants is the modules, sorted, whose manifest claims the seat.
+func (c *Composer) claimants(ctx context.Context, r reader, seat string) []string {
+	c.mu.Lock()
+	recs := maps.Clone(c.modules)
+	c.mu.Unlock()
+	var out []string
+	for _, module := range slices.Sorted(maps.Keys(recs)) {
+		if recs[module].Manifest == "" {
+			continue
+		}
+		raw, err := r.Manifest(ctx, recs[module].Manifest)
+		if err != nil {
+			continue
+		}
+		var man struct {
+			Claims []struct {
+				Name string `json:"name"`
+			} `json:"claims"`
+		}
+		if json.Unmarshal(raw, &man) != nil {
+			continue
+		}
+		for _, cl := range man.Claims {
+			if cl.Name == seat {
+				out = append(out, module)
+				break
+			}
+		}
+	}
+	return out
+}
+
 // seatsOf is the seats a module claims, invokes or provides, from the manifest its module record names; none when it
 // cannot be read, which the reader judges as a module no bootstrapped seat names.
 func (c *Composer) seatsOf(ctx context.Context, r reader, module string) []string {

internal/selfcompose/selfcompose_test.go

@@ -16,9 +16,11 @@ import (
 	"git.novox.be/novox/mesh-delivery/contracts/go/clients/meshcatalogclient"
 	"git.novox.be/novox/mesh-delivery/contracts/go/seats/meshcatalog"
 	"git.novox.be/novox/mesh-delivery/modules/mesh-delivery-builder/builds"
+	"git.novox.be/novox/mesh-sdk/go/declaration"
 	"github.com/novox/mesh-host/internal/buckets"
 	"github.com/novox/mesh-host/internal/buckets/bucketstest"
 	"github.com/novox/mesh-host/internal/compositions"
+	"maps"
 	"os"
 	"reflect"
 	"slices"
@@ -1849,3 +1851,208 @@ func TestAnUnreadSeatListHoldsWhatIsNewAndHandsItAtTheRead(t *testing.T) {
 		t.Errorf("lemurs is still held after the read: %q", still)
 	}
 }
+
+func TestAHeldOwnCompositionThatWasHandedIsNotDropped(t *testing.T) {
+	b := newFakeBus()
+	fp := meshcatalogclient.SourcePrefix2 + strings.Repeat("a", 64)
+	digest := meshcatalogclient.ManifestDigest([]byte(lemurs))
+	man, _ := meshcatalogclient.EncodeManifest(meshcatalog.ManifestsByDigestValue{Module: "lemurs", Manifest: lemurs})
+	key, _ := meshcatalogclient.ManifestsKey(digest)
+	b.kv[buckets.Manifests+"/"+key] = man
+	ctx, cancel := context.WithCancel(context.Background())
+	defer cancel()
+	env := b.env()
+	g := &gate{}
+	changed := make(chan struct{}, 1)
+	env.SeatList, env.SeatListChanged = g, changed
+	passes := make(chan map[string]string, 8)
+	env.Refused = func(_ string, held map[string]string) { passes <- maps.Clone(held) }
+	c := New(env)
+	out, err := c.Watch(ctx, "n1.>")
+	if err != nil {
+		t.Fatal(err)
+	}
+	place, _ := meshmoduleassignerclient.EncodePlacement(meshmoduleassigner.Placement{Node: "n1", Decided: meshmoduleassigner.NodeDecided{Hub: true, Address: "10.77.0.1"},
+		Modules: []meshmoduleassigner.PlacedModule{{Module: "lemurs"}}})
+	have, _ := nodeengineclient.EncodeHaveNode(nodeengine.HaveByNodeNodeValue{Record: &nodeengine.NodeRecord{Node: "n1", TunnelKey: "k", SealingKey: "sk", Capabilities: []string{"systemd"}}})
+	mod, _ := meshcatalogclient.EncodeModule(meshcatalog.ModulesByModuleValue{Fp2: fp, Manifest: digest, Repository: "novox/x", Commit: "c"})
+	want, _ := bucketstest.Want(meshwants.WantsByWantNodeModuleValue{Fp: fp, Self: true})
+	nodeWant, _ := bucketstest.NodeWant(meshwants.WantsByWantNodeNodeValue{Self: true})
+	b.put(t, buckets.Placement, "placement.n1", place)
+	b.put(t, buckets.Have, "n1.node", have)
+	b.put(t, buckets.Modules, "lemurs", mod)
+	b.put(t, buckets.Wants, "want.n1.lemurs", want)
+	b.put(t, buckets.Wants, "want.n1.node", nodeWant)
+	for _, bucket := range []string{buckets.Wants, buckets.Placement, buckets.Have, buckets.Modules, ResolvedBucket, builds.BuilderBuildsBucket} {
+		b.send(bucket, agent.Update{Caught: true})
+	}
+	// The list is read: the node's own composition is handed. Then the reader cannot act again (it restarted and has
+	// not read): the composition is held, and what was handed of it stays, as a held module's does.
+	g.open.Store(true)
+	changed <- struct{}{}
+	for u := next(t, out); !(u.Key == "n1.node" && !u.Deleted); u = next(t, out) {
+	}
+	g.open.Store(false)
+	changed <- struct{}{}
+	for h := range passes {
+		if h[agent.NodeModule] == WaitsForTheSeatListOwn {
+			break
+		}
+	}
+	deadline := time.After(100 * time.Millisecond)
+	for {
+		select {
+		case u := <-out:
+			if u.Key == "n1.node" && u.Deleted {
+				t.Fatalf("the held own composition that was handed is deleted, and the node drops what it runs")
+			}
+		case <-deadline:
+			return
+		}
+	}
+}
+
+// ADR 0636 §1, the cycle of the node's own composition: while the list is unread the node's own composition is held, but
+// a node whose new own composition must start the holder of the seat mesh-catalog composes of it, whatever the list says,
+// the holder's resources and its account and nothing else, over what the node's own composition ran before; and the rest
+// of the new one is handed once the list was read.
+func TestAnUnreadSeatListLetsThroughOfTheOwnCompositionOnlyWhatTheHolderOfTheCatalogNeeds(t *testing.T) {
+	catalog := `{"module":"catalog","version":"1","claims":[{"name":"` + meshcatalog.Seat + `"}]}`
+	other := `{"module":"other","version":"1","claims":[{"name":"node-files"}]}`
+	c := New(newFakeBus().env())
+	c.env.SeatList = &gate{}
+	c.modules = map[string]meshcatalog.ModulesByModuleValue{"catalog": {Manifest: "dc"}, "other": {Manifest: "do"}}
+	c.manifests = map[string][]byte{"dc": []byte(catalog), "do": []byte(other)}
+	res := func(id string) json.RawMessage {
+		return json.RawMessage(`{"id":"` + id + `","type":"directory","path":"/x/` + id + `"}`)
+	}
+	ran := compositions.Composed{Node: "n1", Resources: []json.RawMessage{res("base.dir")},
+		Own: compositions.NodeOwn{Identities: []declaration.Identity{{Module: "base", Name: "mesh-base", UID: 1}}}}
+	next := compositions.Composed{Node: "n1", Resources: []json.RawMessage{res("base.dir"), res("catalog.dir"), res("other.dir")},
+		Own: compositions.NodeOwn{Identities: []declaration.Identity{{Module: "base", Name: "mesh-base", UID: 1},
+			{Module: "catalog", Name: "mesh-catalog", UID: 2}, {Module: "other", Name: "mesh-other", UID: 3}}}}
+	comps := map[string]compositions.Composed{"n1.node": next}
+	held := c.holdUntilListed(context.Background(), reader{c: c}, comps, nil, []compositions.Composed{ran})
+	got, ok := comps["n1.node"]
+	if !ok {
+		t.Fatalf("the node's own composition is held whole, so the holder of the seat mesh-catalog is never started and the list never read")
+	}
+	var ids []string
+	for _, raw := range got.Resources {
+		var h struct{ ID string }
+		_ = json.Unmarshal(raw, &h)
+		ids = append(ids, h.ID)
+	}
+	var names []string
+	for _, i := range got.Own.Identities {
+		names = append(names, i.Name)
+	}
+	if !slices.Equal(ids, []string{"base.dir", "catalog.dir"}) || !slices.Equal(names, []string{"mesh-base", "mesh-catalog"}) {
+		t.Errorf("the own composition lets through resources %v and accounts %v, want what ran and the holder's alone", ids, names)
+	}
+	if held[agent.NodeModule] != WaitsForTheSeatListOwnPart {
+		t.Errorf("the rest of the own composition is held with %q, want %q", held[agent.NodeModule], WaitsForTheSeatListOwnPart)
+	}
+	// Nothing ran: a fresh node composes the holder's part alone.
+	comps = map[string]compositions.Composed{"n1.node": next}
+	c.holdUntilListed(context.Background(), reader{c: c}, comps, nil, nil)
+	if got := comps["n1.node"]; len(got.Resources) != 1 || len(got.Own.Identities) != 1 || got.Own.Identities[0].Name != "mesh-catalog" {
+		t.Errorf("a fresh node lets through %d resources and %d accounts, want the holder's one each", len(got.Resources), len(got.Own.Identities))
+	}
+}
+
+// ADR 0636 §1, the whole case: a node that runs two modules inside its own composition (a build of the old form each),
+// one of them the holder of the seat mesh-catalog, and reads no seat list. Its own composition, handed before the read,
+// carries the holder; the read, said as an event, hands the rest, the other module's resources, with no timer.
+func TestTheHolderOfTheCatalogIsStartedFromTheOwnCompositionBeforeTheListIsReadAndTheRestAfter(t *testing.T) {
+	catalog := `{"module":"catalog","version":"1","claims":[{"name":"` + meshcatalog.Seat + `"}],"capabilities":["systemd"],` +
+		`"resources":[{"id":"service","type":"service","unit":"catalog.service","state":"running","boot":"enabled"}]}`
+	other := `{"module":"other","version":"1","capabilities":["systemd"],` +
+		`"resources":[{"id":"service","type":"service","unit":"other.service","state":"running","boot":"enabled"},` +
+		`{"id":"extra","type":"service","unit":"extra.service","state":"running","boot":"enabled"}]}`
+	b := newFakeBus()
+	ctx, cancel := context.WithCancel(context.Background())
+	defer cancel()
+	env := b.env()
+	g := &gate{}
+	changed := make(chan struct{}, 1)
+	env.SeatList, env.SeatListChanged = g, changed
+	env.Running = func() []compositions.Composed {
+		return []compositions.Composed{{Node: "n1", Resources: []json.RawMessage{json.RawMessage(`{"id":"other.service","type":"service","unit":"other.service","state":"running","boot":"enabled"}`)}}}
+	}
+	c := New(env)
+	out, err := c.Watch(ctx, "n1.>")
+	if err != nil {
+		t.Fatal(err)
+	}
+	mods := map[string]string{"catalog": catalog, "other": other}
+	var placed []meshmoduleassigner.PlacedModule
+	for _, m := range []string{"catalog", "other"} {
+		digest := meshcatalogclient.ManifestDigest([]byte(mods[m]))
+		man, _ := meshcatalogclient.EncodeManifest(meshcatalog.ManifestsByDigestValue{Module: m, Manifest: mods[m]})
+		mk, _ := meshcatalogclient.ManifestsKey(digest)
+		b.kv[buckets.Manifests+"/"+mk] = man
+		rec, _ := meshcatalogclient.EncodeModule(meshcatalog.ModulesByModuleValue{Fp2: meshcatalogclient.SourcePrefix2 + strings.Repeat("a", 64), Manifest: digest, Repository: "novox/x", Commit: "c"})
+		b.put(t, buckets.Modules, m, rec)
+		hk, _ := nodeengineclient.HaveKey("n1", m)
+		hv, _ := nodeengineclient.EncodeHave(nodeengine.HaveByNodeModuleValue{Build: "build-1", Phase: nodeengineclient.PhaseRunning, Health: nodeengineclient.HealthOK})
+		b.put(t, buckets.Have, hk, hv)
+		placed = append(placed, meshmoduleassigner.PlacedModule{Module: m})
+	}
+	place, _ := meshmoduleassignerclient.EncodePlacement(meshmoduleassigner.Placement{Node: "n1", Decided: meshmoduleassigner.NodeDecided{Hub: true, Address: "10.77.0.1"}, Modules: placed})
+	have, _ := nodeengineclient.EncodeHaveNode(nodeengine.HaveByNodeNodeValue{Record: &nodeengine.NodeRecord{Node: "n1", TunnelKey: "k", SealingKey: "sk", Capabilities: []string{"systemd"}}})
+	nodeWant, _ := bucketstest.NodeWant(meshwants.WantsByWantNodeNodeValue{Self: true})
+	b.put(t, buckets.Placement, "placement.n1", place)
+	b.put(t, buckets.Have, "n1.node", have)
+	b.put(t, buckets.Wants, "want.n1.node", nodeWant)
+	for _, bucket := range []string{buckets.Wants, buckets.Placement, buckets.Have, buckets.Modules, ResolvedBucket, builds.BuilderBuildsBucket} {
+		b.send(bucket, agent.Update{Caught: true})
+	}
+	servicesIn := func(u agent.Update) []string {
+		var comp compositions.Composed
+		if err := json.Unmarshal(u.Value, &comp); err != nil {
+			t.Fatalf("the own composition does not decode: %v", err)
+		}
+		var ids []string
+		for _, raw := range comp.Resources {
+			var h struct{ ID string }
+			_ = json.Unmarshal(raw, &h)
+			if strings.HasPrefix(h.ID, "catalog.") || strings.HasPrefix(h.ID, "other.") {
+				ids = append(ids, h.ID)
+			}
+		}
+		return ids
+	}
+	var first agent.Update
+	for first.Key == "" {
+		if u := next(t, out); u.Key == "n1.node" && !u.Deleted {
+			first = u
+		}
+	}
+	if got := servicesIn(first); !slices.Equal(got, []string{"other.service", "catalog.service"}) {
+		t.Fatalf("before the read the own composition carries services %v, want what ran and the holder's alone", got)
+	}
+	g.open.Store(true)
+	changed <- struct{}{}
+	for {
+		u := next(t, out)
+		if u.Key == "n1.node" && !u.Deleted {
+			if got := servicesIn(u); !slices.Equal(got, []string{"catalog.service", "other.service", "other.extra"}) {
+				t.Fatalf("after the read the own composition carries services %v, want what ran, the rest of the module and the holder", got)
+			}
+			return
+		}
+	}
+}
+
+// ADR 0636 §1: a reader the node could not start is no reader that gates nothing: the node holds like one that waits.
+func TestACompositionHoldsWhenTheSeatListReaderCouldNotStart(t *testing.T) {
+	gate := Unread{}
+	c := New(newFakeBus().env())
+	comps := map[string]compositions.Composed{"n1.node": {Node: "n1"}}
+	c.env.SeatList = gate
+	held := c.holdUntilListed(context.Background(), reader{c: c}, comps, nil, nil)
+	if _, ok := held[agent.NodeModule]; !ok || len(comps) != 0 {
+		t.Errorf("held %v, comps %v: a reader that did not start must hold the node's own composition", held, comps)
+	}
+}