novox/mesh-delivery
10
0
Fork 0
Code Issues Pull Requests 1 Actions Packages Projects Releases Wiki Activity

GR9: the grants part follows today's catalogue contracts and reads a stopped running as stopped

Browse Source 
SeatsToPassOn takes what each node-engine's running said and a reader of entries, Assignment names the module and its
branch, Hearing takes the defined-events function (nil here: GR4's). HearRunning keeps a stopped running with no
fingerprint and no manifest, and a stopped module claims nothing on its node.

Builds: design 47 G §404, §459, §199, §386 (nil defines none)

Claude-Session: https://claude.ai/code/session_01JG73KEq4zoDfM9WtYpcQgZ
This commit is contained in:
jschoubben
2026-11-11 21:31:10 +01:00
parent f8e509ce56
commit f0d041b8dc
4 changed files with 70 additions and 20 deletions
Download Patch File Download Diff File Expand all files Collapse all files
modules/mesh-grants/grants_test.go
+26 -4
View File
@@ -1343,6 +1343,14 @@ func TestGrants(t *testing.T) {
{"the holder's running build still claims it", map[string]catalogue.Manifest{"a.mesh-catalog": claimer("mesh-catalog")}, nil, []string{"mesh-catalog"}, ""},
{"a second claimant", nil, []string{"other"}, nil, "seat mesh-catalog --to <node>/<module>"},
}
// A holder whose node-engine said `stopped` claims nothing there (design 47 G §404, §459).
rows = append(rows, struct {
name string
running map[string]catalogue.Manifest
extra []string
want []string
said string
}{"the holder said stopped by its node-engine", map[string]catalogue.Manifest{"a.mesh-catalog": claimer("mesh-catalog")}, nil, []string{"mesh-delivery-indexer"}, "composed as held by mesh-delivery-indexer"})
// What the want names is no running build: the seat's holder is read from the digest the node-engine said it runs.
wantedOnly := rows[1]
wantedOnly.name = "a build only wanted, not said running by its node-engine, is not read"
@@ -1359,13 +1367,20 @@ func TestGrants(t *testing.T) {
Holds: []meshmoduleassigner.Holding{{Claim: "mesh-catalog", Scope: "mesh", Module: "mesh-catalog"}}},
"b": {Node: "b", Modules: []meshmoduleassigner.PlacedModule{{Module: "mesh-delivery-indexer"}}}}}
if !strings.Contains(row.name, "only wanted") {
in.Ran = row.running
in.Ran, in.Heard, in.RanByFingerprint = row.running, map[string]catalogue.Heard{}, map[string]catalogue.Manifest{}
for key, m := range row.running {
in.Heard[key] = catalogue.Heard{Fingerprint: "fp-" + key}
in.RanByFingerprint["fp-"+key] = m
}
}
maps.Copy(in.Running, row.running)
for _, c := range row.extra {
in.Manifests[c] = claimer(c)
in.Assigned["b"] = append(in.Assigned["b"], c)
}
if strings.Contains(row.name, "stopped") {
in.Heard["a.mesh-catalog"] = catalogue.Heard{Stopped: true}
}
before := maps.Clone(in.Placed)
for node, p := range before {
p.Holds = slices.Clone(p.Holds)
@@ -2084,8 +2099,9 @@ func TestGrants(t *testing.T) {
kb.loaded = shadowOf(t, kb)
kp := partOn(kb)
kp.Create, kp.Consumers = kb, kb
keepDays, keepMiB := 7, 64
kp.SeatList = keptGate{seatdef.Definition{Seat: "mesh-conditions", Events: []seatdef.Message{
{Name: "condition-raised", Keep: seatdef.KeepStream, MaxAgeDays: 7, MaxMiB: 64}}}}
{Name: "condition-raised", Keep: &seatdef.Keep{Object: true, MaxAgeDays: &keepDays, MaxMiB: &keepMiB}}}}}
kp.Compose()
if !slices.Contains(kb.creates, "SEAT_mesh-conditions_condition-raised") || !slices.Contains(kb.consumerCreates, "SEAT_mesh-conditions_condition-raised.b_mod-y") ||
strings.Contains(fmt.Sprint(kb.consumers["EVENTS"]), "condition-raised") ||
@@ -4776,6 +4792,10 @@ func TestTheGrantsPartKeepsWhatEachNodeRuns(t *testing.T) {
raw, _ := json.Marshal(nodeengine.RunningEvent{V: 1, At: time.Now(), Fingerprint: fp, Manifest: digest, Module: module, Node: node})
return raw
}
stopped := func(node, module string) []byte {
raw, _ := json.Marshal(nodeengine.RunningEvent{V: 1, At: time.Now(), Module: module, Node: node, Stopped: true})
return raw
}
bus := newFakeBus()
p := &grants.Part{Store: ownStore{bus}}
subject := nodeengine.SubjectRunning("ace", "mod-c")
@@ -4790,6 +4810,8 @@ func TestTheGrantsPartKeepsWhatEachNodeRuns(t *testing.T) {
{"a new build is kept", subject, body("ace", "mod-c", "fp2", "sha256:b"), true, ""},
{"a body of another node than its subject is refused", subject, body("g14", "mod-c", "fp2", "sha256:b"), false, "does not carry"},
{"a body with no manifest is refused", subject, body("ace", "mod-c", "fp2", ""), false, "no node, module or manifest"},
{"a stopped body carries no fingerprint or manifest and is kept", subject, stopped("ace", "mod-c"), true, ""},
{"the same stopped said again keeps nothing", subject, stopped("ace", "mod-c"), false, ""},
{"a body that is no running is refused", subject, []byte("x"), false, "not a running body"},
} {
t.Run(row.name, func(t *testing.T) {
@@ -4804,7 +4826,7 @@ func TestTheGrantsPartKeepsWhatEachNodeRuns(t *testing.T) {
t.Fatal(err)
}
said := got.(grants.RunningSaid).Running
if len(said) != 1 || said[0].Manifest != "sha256:b" {
t.Errorf("grants_running = %+v, want the last build of ace.mod-c", said)
if len(said) != 1 || !said[0].Stopped {
t.Errorf("grants_running = %+v, want the last word of ace.mod-c: stopped", said)
}
}
modules/mesh-grants/internal/grants/inputs.go
+4
View File
@@ -158,6 +158,10 @@ type Inputs struct {
// Ran is the manifest of the digest each node-engine says it runs, by `<node>.<module>` (running.go): what a seat's
// holder is read from. A node that said nothing has none, and its module is read from the trunk's manifest.
Ran map[string]catalogue.Manifest
// Heard is what each node-engine's `running` said of each `<node>.<module>`: a fingerprint, or stopped (design 47 G §404,
// §459); a key absent was not heard since the part's consumer began. RanByFingerprint is the manifest of each fingerprint heard.
Heard map[string]catalogue.Heard
RanByFingerprint map[string]catalogue.Manifest
// Had is the manifest of the build each machine runs now (its have's fingerprint, the builder's build of it), by
// `<node>.<module>`; the want's Running may already be a newer build. A membership is composed from both and the trunk's.
Had map[string]catalogue.Manifest
modules/mesh-grants/internal/grants/records.go
+24 -13
View File
@@ -426,7 +426,7 @@ func hearsOf(m catalogue.Manifest, declared map[string]catalogue.Manifest, node
consumes []string) ([]buscompose.Heard, bool, []string) {
reading := m
reading.Consumes = consumes
parts, touches, onEvents := catalogue.Hearing(reading, declared, node, values)
parts, touches, onEvents := catalogue.Hearing(reading, declared, node, values, nil)
var out []buscompose.Heard
for _, p := range parts {
h := buscompose.Heard{Stream: p.Stream.BusName(), Emitter: p.Emitter, Restricted: p.Stream.Grant != nil}
@@ -456,20 +456,14 @@ func holdersAfterPass(in Inputs) ([]catalogue.Held, []Gap) {
var assignments []catalogue.Assignment
for _, node := range slices.Sorted(maps.Keys(in.Assigned)) {
for _, module := range in.Assigned[node] {
m, ok := in.Ran[node+"."+module]
if !ok {
m, ok = in.Manifests[module]
}
if !ok {
m, ok = in.Held[module]
}
if !ok {
m = catalogue.Manifest{Module: module}
}
assignments = append(assignments, catalogue.Assignment{Node: node, Manifest: m, Unread: !ok})
assignments = append(assignments, catalogue.Assignment{Node: node, Module: module})
}
}
passes, said := catalogue.SeatsToPassOn(held, assignments)
heard := func(node, module string) (catalogue.Heard, bool) {
h, ok := in.Heard[node+"."+module]
return h, ok
}
passes, said := catalogue.SeatsToPassOn(held, assignments, heard, ranEntries{in})
var gaps []Gap
for _, p := range passes {
held = slices.DeleteFunc(held, func(h catalogue.Held) bool { return h.Kind == p.From.Kind && sameSeat(h.Claim, p.Seat.Name) })
@@ -485,6 +479,23 @@ func holdersAfterPass(in Inputs) ([]catalogue.Held, []Gap) {
return held, gaps
}
// ranEntries is the reader of entries SeatsToPassOn takes (design 47 G §459): the manifest of a fingerprint a node-engine
// said it runs, and a module's manifest where none was heard: the trunk's, else the running build of a held module.
type ranEntries struct{ in Inputs }
func (r ranEntries) ByFingerprint(fingerprint string) (catalogue.Manifest, bool) {
m, ok := r.in.RanByFingerprint[fingerprint]
return m, ok
}
func (r ranEntries) OnBranch(module, _ string) (catalogue.Manifest, bool) {
if m, ok := r.in.Manifests[module]; ok {
return m, true
}
m, ok := r.in.Held[module]
return m, ok
}
// heldWherePlaced is a module's holds on a node as the catalogue's assignments says them (ADR 0549 §1, §3). The assignments'
// holds are the holders on record, as catalogue.World's Holdings are, after the passes of holdersAfterPass: a seat held
// once for the mesh that some node's assignments holds is held only where assignments names it (this node's naming this
modules/mesh-grants/internal/grants/running.go
+16 -3
View File
@@ -30,7 +30,8 @@ func (p *Part) HearRunning(subject string, data []byte) (changed bool, err error
if err := json.Unmarshal(data, &r); err != nil {
return false, fmt.Errorf("%s is not a running body: %w", subject, err)
}
if r.Node == "" || r.Module == "" || r.Manifest == "" {
// A `running` with stopped carries no fingerprint and no manifest (design 47 G §404, §365): the have no longer holds the module.
if r.Node == "" || r.Module == "" || (r.Manifest == "" && !r.Stopped) {
return false, fmt.Errorf("%s names no node, module or manifest digest", subject)
}
if !strings.HasSuffix(subject, "."+RunningKey(r.Node, r.Module)) {
@@ -43,7 +44,7 @@ func (p *Part) HearRunning(subject string, data []byte) (changed bool, err error
}
if found {
var w nodeengine.RunningEvent
if json.Unmarshal(was, &w) == nil && w.Manifest == r.Manifest && w.Fingerprint == r.Fingerprint {
if json.Unmarshal(was, &w) == nil && w.Manifest == r.Manifest && w.Fingerprint == r.Fingerprint && w.Stopped == r.Stopped {
return false, nil
}
}
@@ -57,6 +58,8 @@ func (p *Part) HearRunning(subject string, data []byte) (changed bool, err error
// whose manifest cannot be read is a gap, and that node's module is read as it was before (the trunk's).
func (p *Part) readRan(in *Inputs) {
in.Ran = map[string]catalogue.Manifest{}
in.Heard = map[string]catalogue.Heard{}
in.RanByFingerprint = map[string]catalogue.Manifest{}
keys, err := p.Store.Keys(StateRunning)
if err != nil {
in.Gaps = append(in.Gaps, Gap{Input: StateRunning, Why: "its own state cannot be read, so no seat is read from what a node runs: " + err.Error()})
@@ -69,7 +72,15 @@ func (p *Part) readRan(in *Inputs) {
continue
}
var r nodeengine.RunningEvent
if json.Unmarshal(raw, &r) != nil || r.Manifest == "" {
if json.Unmarshal(raw, &r) != nil {
continue
}
if r.Stopped {
// A stopped module is heard as stopped and claims nothing on that node (design 47 G §404, §459).
in.Heard[key] = catalogue.Heard{Stopped: true}
continue
}
if r.Manifest == "" {
continue
}
m, err := p.cache.get(p.Read, r.Manifest)
@@ -78,6 +89,8 @@ func (p *Part) readRan(in *Inputs) {
continue
}
in.Ran[key] = m
in.Heard[key] = catalogue.Heard{Fingerprint: r.Fingerprint}
in.RanByFingerprint[r.Fingerprint] = m
}
}