Merge branch 'b7i/registry' into b7/registry

This commit is contained in:
2026-11-05 16:30:27 +01:00
15 changed files with 225 additions and 14 deletions
@@ -461,10 +461,10 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro
return nil
}
reader.Start(ctx, js, cred, process.AwaitBucket(ctx, js, conn, firstViewsBucket, "checker_first_views reads nothing and checker_first_view_again is not heard", logger.Printf))
// Every part has registered. Where the consumer is not there yet no package registry's published is heard and a group's
// check fails at its publish bound until the checker starts again.
if err := hub.Start(ctx, js, cred); err != nil {
logger.Printf("%v, so no package registry's published is heard and a group's check fails at its publish bound until the checker starts again", err)
// Every part has registered. Where the consumer is not there yet the hub binds it as soon as it is, and the seat list is read
// again then if it is still unread.
if err := hub.StartUntilBound(ctx, js, cred, nil, func() { readIfUnread(seats.reader) }); err != nil {
logger.Printf("%v, so no package registry's published is heard until the consumer is there; the checker tries again", err)
}
defer hub.Stop()
// The checks, the tries and the hand asks, once every first view is answered (the lead's ruling for DLV-M2): a check
@@ -49,3 +49,12 @@ func (s seatList) Definitions() ([]json.RawMessage, error) {
}
return out, nil
}
// readIfUnread is called when the checker's hub is heard. The seat may have answered, and said that it does, before the
// checker's consumer existed, whose events start at the stream's end: the first read found it silent and nothing else reads
// again until its next event, so while the list is unread it is read now, on a goroutine of its own (ADR 0636 §1).
func readIfUnread(reader *seatlist.Reader) {
if reader != nil {
reader.ReloadIfUnread()
}
}
@@ -0,0 +1,41 @@
package main
import (
"encoding/json"
"errors"
"sync/atomic"
"testing"
"time"
"git.novox.be/novox/mesh-sdk/go/seatlist"
)
// silentThenAnswering is a seat that is silent at the reader's first read and answers after.
type silentThenAnswering struct{ answering atomic.Bool }
func (b *silentThenAnswering) Ask(string, any) (json.RawMessage, error) {
if !b.answering.Load() {
return nil, errors.New("not connected yet")
}
return json.RawMessage(`{"revision":1,"seats":[]}`), nil
}
func (b *silentThenAnswering) Subscribe(string, func(string, json.RawMessage) error) error { return nil }
// A seat that spoke before the checker's consumer existed is read when the hub is heard (ADR 0636 §1).
func TestTheSeatListIsReadWhenTheHubIsHeard(t *testing.T) {
bus := &silentThenAnswering{}
reader := followSeatList(bus, t.Logf)
<-reader.Ready()
if reader.Status().Heard {
t.Fatal("a silent seat was heard")
}
bus.answering.Store(true)
readIfUnread(reader)
for deadline := time.Now().Add(5 * time.Second); !reader.Status().Heard; time.Sleep(10 * time.Millisecond) {
if time.Now().After(deadline) {
t.Fatal("the list was not read when the hub was heard")
}
}
var _ seatlist.Bus = bus
}
@@ -408,7 +408,8 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro
seatDefs := process.AwaitBucket(ctx, js, conn, index.SeatDefinitionsBucket, "no seat definition is kept", logger.Printf)
marks := process.AwaitBucket(ctx, js, conn, index.MarksBucket, "no change is said on the views family, so every reader waits on its first view", logger.Printf)
seatBus := &seatBus{defs: awaitedSource{ctx, seatDefs}, marks: awaitedSource{ctx, marks}}
_, saySeats := followSeatList(seatBus, saySeatDefinition(js, cred), logger.Printf)
seatReader, saySeats := followSeatList(seatBus, saySeatDefinition(js, cred), logger.Printf)
readWhenBound(seatReader, seatDefs, marks)
ix := &index.Indexer{Findings: awaitedStore{findings}, OldSections: awaitedStore{oldSections}, Requirements: awaitedStore{requirements}, Store: kvStore{candidates}, Modules: kvStore{modules}, Manifests: kvStore{manifests}, Taken: awaitedStore{taken}, Builds: known,
Verdicts: awaitedStore{verdicts}, SayRefused: sayRefused(js, cred, logger.Printf), SayAccepted: sayAccepted(js, cred),
Predefined: awaitedStore{predefined}, ConfigCandidates: awaitedStore{configCandidates}, Logf: logger.Printf,
@@ -129,3 +129,20 @@ func followSeatList(bus *seatBus, say func(context.Context, index.SeatDefinition
return nil
}
}
// bucket is a state bound when it appears (process.AwaitedBucket).
type bucket interface{ Ready(func(jetstream.KeyValue)) }
// readWhenBound has the reader read the list again as each of the states it is answered from is bound. The reader starts before
// they are, its first read finds them missing and leaves it unread, and the indexer's own events (said) are the only other thing
// that reads it again: a trunk that moves no definition after a start would leave the list unread for good (ADR 0636 §1).
func readWhenBound(reader *seatlist.Reader, states ...bucket) {
if reader == nil {
return
}
for _, state := range states {
state.Ready(func(jetstream.KeyValue) {
reader.ReloadIfUnread()
})
}
}
@@ -0,0 +1,66 @@
package main
import (
"errors"
"sync"
"testing"
"time"
"github.com/nats-io/nats.go/jetstream"
"git.novox.be/novox/mesh-sdk/go/seatlist"
"git.novox.be/novox/mesh-sdk/go/views"
)
type unboundSource struct {
mu sync.Mutex
bound bool
}
func (s *unboundSource) Keys() ([]string, error) {
s.mu.Lock()
defer s.mu.Unlock()
if !s.bound {
return nil, errors.New("not yet")
}
return nil, nil
}
func (s *unboundSource) Get(string) (*views.Entry, bool, error) {
s.mu.Lock()
defer s.mu.Unlock()
if !s.bound {
return nil, false, errors.New("not yet")
}
return nil, false, nil
}
type awaitedFake struct{ ready []func(jetstream.KeyValue) }
func (b *awaitedFake) Ready(f func(jetstream.KeyValue)) { b.ready = append(b.ready, f) }
// The indexer's reader of its own seat list starts before its states are bound, so its first read finds them missing; it reads the
// list again when they are bound, with no definition moving (ADR 0636 §1).
func TestTheSeatListIsReadAgainWhenItsStatesAreBound(t *testing.T) {
src := &unboundSource{}
bus := &seatBus{defs: src, marks: src}
reader, _ := followSeatList(bus, nil, t.Logf)
<-reader.Ready()
if reader.Status().Heard {
t.Fatal("the list was read with its states missing")
}
b := &awaitedFake{}
readWhenBound(reader, b)
src.mu.Lock()
src.bound = true
src.mu.Unlock()
for _, f := range b.ready {
f(nil)
}
for deadline := time.Now().Add(5 * time.Second); !reader.Status().Heard; time.Sleep(10 * time.Millisecond) {
if time.Now().After(deadline) {
t.Fatal("the list was not read after its states were bound")
}
}
var _ seatlist.Bus = bus
}
@@ -193,7 +193,7 @@ func TestTheGrantsPartWhole(t *testing.T) {
// has heard the seat and holds the bootstrapped definitions, and every module is composed. (The unit case holds a module
// while the list is unread.)
var seatAnswering atomic.Bool
seatAnswering.Store(true)
seatAnswering.Store(false)
for _, subject := range []string{"mesh.seat.mesh-catalog.call.seat-definitions", "mesh.seat.mesh-catalog.call.seat-definitions.>"} {
if _, err := nc.Subscribe(subject, func(m *nats.Msg) {
if !seatAnswering.Load() {
@@ -239,6 +239,14 @@ func TestTheGrantsPartWhole(t *testing.T) {
if _, err := kvs[grants.Module+"."+grants.StateMemberships].Get(ctx, "a.mod-c"); err == nil {
t.Error("a membership composed before the owners answered its first views")
}
// ADR 0636 §1, the unread phase: the seat was silent when the part started and speaks now, and says so on its event
// before the part's own consumer exists (the part makes it from its first composition, which waits on its owners'
// first views). The event is lost to that consumer, which starts at the stream's end, so the part reads the list again
// when its consumer is heard; mod-c, which the unread list held, is composed after that and not before.
seatAnswering.Store(true)
if _, err := js.Publish(ctx, "mesh.seat.mesh-catalog.event.seat-definition-changed", []byte(`{"seat":"mesh-broker","source":"trunk"}`)); err != nil {
t.Fatal(err)
}
answering.Store(true)
until("a.mod-c's membership in the part's own state memberships", func() bool {
e, err := kvs[grants.Module+"."+grants.StateMemberships].Get(ctx, "a.mod-c")
@@ -183,6 +183,7 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro
if err := hub.Start(ctx, js, account); err != nil {
return err
}
gate.eventsLive()
hears, err := grants.OwnHears(ownStore{conn}, account.Node)
if err != nil {
return err
@@ -49,6 +49,16 @@ func (g *seatGate) follow(bus seatlist.Bus, changed func(), logf func(string, ..
return reader
}
// eventsLive is called when the part's own consumer on EVENTS is heard. A seat that was silent at the part's start may have said it
// speaks before that consumer existed, which starts at the stream's end, so its event is never heard: while the list is unread the
// part reads it now (ADR 0636 §1), on a goroutine of its own so the pass is not held; the reader changes the catalogue and wakes
// the part when the seat answers, and stays unread, to be read at the seat's next event, when it does not.
func (g *seatGate) eventsLive() {
if r := g.reader.Load(); r != nil {
r.ReloadIfUnread()
}
}
// tools are the two tools over whatever reader the gate holds at the call.
func (g *seatGate) tools() []stdio.Tool {
names := []string{grants.ToolSeatList, grants.ToolSeatListAgain}
+31 -2
View File
@@ -1274,8 +1274,10 @@ func TestGrants(t *testing.T) {
n.manifestOf("node-tools", `{"module":"node-tools","version":"1","tool-runner":true}`) // the machine's tool runner, which carries the seat
n.assign("novox", "node-tools")
// The live placement names mesh-host; with its manifest's state the runner once carried it and the whole list was
// refused as a second writer of the engine's health (issue 774, 2026-10-30): the shadow below now composes.
n.manifestOf(buscompose.EngineModule, `{"module":"mesh-host","version":"1","consumes":["restic.backed-up"],"state":[{"name":"health"}]}`)
// refused as a second writer of the engine's health (issue 774, 2026-10-30): the shadow below now composes, with the states the
// engine's manifest declares today (health, findings, data, packages): the findings state, written by each node's engine
// alone (ADR 0636 §1), does not bring the refusal back, and the runner is granted none of them.
n.manifestOf(buscompose.EngineModule, `{"module":"mesh-host","version":"1","consumes":["restic.backed-up"],"state":[{"name":"health"},{"name":"findings"},{"name":"data"},{"name":"packages"}]}`)
n.put(grants.ReadHave, "novox.node", map[string]any{"v": 1})
pl, _ = delivery.EncodePlacement(seed)
n.states[grants.ReadPlacement] = map[string][]byte{"placement.novox": pl}
@@ -2256,6 +2258,33 @@ func TestGrants(t *testing.T) {
}
})
// What the part needs to hear the seat and read the list is never held by the unread list (ADR 0636 §1; the lead's ruling
// of 2026-11-05): its own module (its consumer on EVENTS, its grant of the seat's verb) and the module that holds the seat
// the list is read through, whatever other seats their manifests name.
t.Run("the seat list unread never holds the part's own module or the holder of the seat it is read through", func(t *testing.T) {
b := fixture()
own, err := os.ReadFile("module.json")
if err != nil {
t.Fatal(err)
}
b.manifestOf(grants.Module, string(own))
b.manifestOf("cat-holder", `{"module":"cat-holder","version":"1","claims":[{"name":"mesh-catalog"}],"invokes":["seat:mesh-vault.account-issue"]}`)
b.assign("a", grants.Module)
b.assign("a", "cat-holder")
p := partOn(b)
p.SeatList = &seatGate{} // nothing is bootstrapped: only the exemption lets these two through
p.Compose()
refused := strings.Join(p.Last().Refused, "\n")
for _, module := range []string{grants.Module, "cat-holder"} {
if strings.Contains(refused, "manifest of "+module+":") {
t.Errorf("%s is held while the seat list is unread: %v", module, p.Last().Refused)
}
}
if !strings.Contains(refused, "manifest of mod-x: it waits for the seat list") {
t.Errorf("another module is not held: %v", p.Last().Refused)
}
})
// DLV-G1, 1966/AC32-AC33 (novox/hq ADR 0595 §1, §7; design 25 § 12.2): the bus module's created writes the binding into
// the module's membership on every machine it is assigned, one assigned later included; recreated rewrites it; dropped
// removes it; before created, on a refused, or from an emitter holding no mesh-broker, none.
@@ -816,6 +816,11 @@ func healthChecks(t *testing.T) {
MayPublish(perms, "$JS.API.DIRECT.GET.KV_mesh-controller_declared-packages.$KV.mesh-controller_declared-packages.a")) {
t.Errorf("node %s writes the packages keys %v", u.Node, perms.Publish)
}
// ADR 0564 §2, ADR 0636 §1: and the findings it raises about its own node, its own node's keys alone.
if u.Kind == KindNode && (!MayPublish(perms, "$KV.mesh-host_findings.machine.a.seat-list-unread") || MayPublish(perms, "$KV.mesh-host_findings.machine.b.seat-list-unread") ||
MayPublish(perms, "$KV.mesh-host_findings.a")) {
t.Errorf("node %s writes the findings keys %v", u.Node, perms.Publish)
}
if u.Kind == KindModule && MayPublish(perms, "$KV.mesh-delivery-assigner_policies.x") {
t.Errorf("%s writes the assigner's policies", u.Username())
}
@@ -32,6 +32,17 @@ var EnginePackagesBucket = BucketName(EngineModule, delivery.EnginePackagesState
// key `<node>.<module>` per module, written by that node's engine alone.
var EngineDataBucket = delivery.EngineDataBucket
// EngineFindingsBucket is the engine's state of the findings it raises about its own node (novox/hq ADR 0564 §2, ADR 0636 §1):
// keys `machine.<node>.<kind>`, written by that node's engine alone and cleared by it; the condition keeper raises each as a
// condition.
var EngineFindingsBucket = BucketName(EngineModule, "findings")
// EngineFindingsGrants is what a node's engine publishes to write and clear its node's findings: its own node's keys alone,
// and their read. Without it the finding that the node waits for the seat list is refused on the bus and never raised.
func EngineFindingsGrants(node string) []string {
return append([]string{"$KV." + EngineFindingsBucket + ".machine." + node + ".>"}, readGrants(EngineFindingsBucket, "machine."+node+".>")...)
}
// EnginePackagesGrants is what a node's engine publishes to keep its node's key of the packages bucket: the key alone,
// and its read.
func EnginePackagesGrants(node string) []string {
@@ -80,5 +91,9 @@ func engineGrants(p Principal) (pub, sub []string) {
if safeSubject.MatchString(p.Node) {
pub = append(pub, EnginePackagesGrants(p.Node)...)
}
// And the findings it raises about its own node, its own node's keys alone (ADR 0564 §2, ADR 0636 §1).
if safeSubject.MatchString(p.Node) {
pub = append(pub, EngineFindingsGrants(p.Node)...)
}
return append(pub, tp...), append(sub, ts...)
}
File diff suppressed because one or more lines are too long
File diff suppressed because one or more lines are too long
+10 -1
View File
@@ -17,6 +17,7 @@ import (
"sync"
"time"
"git.novox.be/novox/mesh-delivery/contracts/go/seats/meshcatalog"
"git.novox.be/novox/mesh-sdk/go/catalogue"
"git.novox.be/novox/mesh-sdk/go/delivery"
)
@@ -219,6 +220,14 @@ type SeatGate interface {
MayAct(seats ...string) bool
}
// readsTheList says whether a module is what the part needs to hear the seat and read the list: the part's own module (its
// consumer on EVENTS, its grant of the seat's verb and events) and the holder of the seat the list is read through. The unread
// list never holds these, whatever other seats their manifests name: held, the seat's event could not reach the part and
// the list would never be read (ADR 0636 §1).
func readsTheList(module string, m catalogue.Manifest) bool {
return module == Module || slices.ContainsFunc(m.Claims, func(c catalogue.Claim) bool { return c.Name == meshcatalog.Seat })
}
// SeatsOf are the seats a manifest claims or invokes, which the seat list must name before the part composes the module.
func SeatsOf(m catalogue.Manifest) []string {
var seats []string
@@ -422,7 +431,7 @@ func ReadInputs(r Reader, cache *Manifests) Inputs {
if err == nil {
m, err = cache.get(r, v.Manifest)
}
if err == nil && gate != nil && !gate.MayAct(SeatsOf(m)...) {
if err == nil && gate != nil && !readsTheList(module, m) && !gate.MayAct(SeatsOf(m)...) {
err = errors.New("it waits for the seat list: the list has not been read yet and the manifest names seats the bootstrapped list does not (novox/hq ADR 0636 §1)")
}
if err != nil {