diff --git a/modules/mesh-grants/grants_test.go b/modules/mesh-grants/grants_test.go index cf70a3e7..380e99ae 100644 --- a/modules/mesh-grants/grants_test.go +++ b/modules/mesh-grants/grants_test.go @@ -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 /"}, } + // 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) } } diff --git a/modules/mesh-grants/internal/grants/inputs.go b/modules/mesh-grants/internal/grants/inputs.go index 4904bb14..5f6fc1e4 100644 --- a/modules/mesh-grants/internal/grants/inputs.go +++ b/modules/mesh-grants/internal/grants/inputs.go @@ -158,6 +158,10 @@ type Inputs struct { // Ran is the manifest of the digest each node-engine says it runs, by `.` (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 `.`: 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 // `.`; 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 diff --git a/modules/mesh-grants/internal/grants/records.go b/modules/mesh-grants/internal/grants/records.go index c20ef223..bb06a96b 100644 --- a/modules/mesh-grants/internal/grants/records.go +++ b/modules/mesh-grants/internal/grants/records.go @@ -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 diff --git a/modules/mesh-grants/internal/grants/running.go b/modules/mesh-grants/internal/grants/running.go index 8cda4a51..6cf999c8 100644 --- a/modules/mesh-grants/internal/grants/running.go +++ b/modules/mesh-grants/internal/grants/running.go @@ -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 } }