Assigner: hears the builder's build log and writes a kept build's parts and keeping node into the want

The want names parts (by <module>:<artifact>) and kept-on from the build's end, only for a node whose engine states want-parts.

Builds: design 47 G §526, §379, §525, §671
This commit is contained in:
2026-11-11 23:26:52 +01:00
parent a245c2f1e9
commit 457b08ce15
10 changed files with 199 additions and 6 deletions
@@ -0,0 +1,84 @@
package main
import (
"fmt"
"sync"
stdio "git.novox.be/novox/mesh-sdk/go"
"git.novox.be/novox/mesh-sdk/go/process"
"git.novox.be/novox/mesh-delivery/modules/mesh-delivery-builder/builds"
"mesh-delivery-assigner/internal/assign"
)
// The builder's log, heard (design 47 G §379, §526, §671): each `build.ended` of a done build is kept by module and fingerprint,
// so a want names a kept build's parts by digest and the node that keeps it, taken from the build's end and never from the
// builder's done key (ADR 0596 §2). The log is read on the assigner's own credential as a read of `mesh-delivery-builder.build-log`
// (its manifest's `reads`), every entry from the start of the log and then each new one, and wakes the pass on each.
// builtEnds are the done builds' ends heard, by module and fingerprint (putBuilt).
type builtEnds struct {
mu sync.Mutex
ends map[string]builds.BuildLogEntry
}
func (b *builtEnds) key(module, fp string) string { return module + "@" + fp }
// putBuilt takes one entry of the builder's log: a done build's end is kept, anything else is passed over.
func (b *builtEnds) putBuilt(value []byte) error {
if !builds.IsBuildEntry(value) {
return nil
}
e, err := builds.DecodeBuildLogEntry(value)
if err != nil {
return err
}
if e.Event != builds.BuildEndedEvent || e.Outcome != builds.BuildDone {
return nil
}
if e.Module == "" || e.FP == "" {
return fmt.Errorf("a build's end naming no module or fingerprint: %s", value)
}
b.mu.Lock()
defer b.mu.Unlock()
if b.ends == nil {
b.ends = map[string]builds.BuildLogEntry{}
}
if old, ok := b.ends[b.key(e.Module, e.FP)]; !ok || !e.At.Before(old.At) {
b.ends[b.key(e.Module, e.FP)] = e
}
return nil
}
// keptBuild is the kept build of module at fp, as its end names it: each artifact's digest by `<module>:<artifact>`, and the node
// that keeps it. False for a build not done or not kept (it names no keeping node).
func (b *builtEnds) keptBuild(module, fp string) (assign.KeptBuild, bool) {
b.mu.Lock()
e, ok := b.ends[b.key(module, fp)]
b.mu.Unlock()
if !ok || e.KeptOn == "" {
return assign.KeptBuild{}, false
}
parts := map[string]string{}
for _, a := range e.Made {
parts[module+":"+a.Name] = a.Digest
}
return assign.KeptBuild{Parts: parts, KeptOn: e.KeptOn}, true
}
// hearBuildLog reads the builder's log on conn and wakes the pass on each entry; the watch lasts as long as the connection.
func hearBuildLog(conn *process.Conn, ends *builtEnds, wake func(), logf func(string, ...any)) {
go func() {
err := conn.ReadLog(builds.BuilderModule, builds.BuilderLog).Watch("", stdio.LogFromAll, func(le stdio.LogEntry) error {
if err := ends.putBuilt(le.Value); err != nil {
logf("%s entry %d is not read: %v", builds.BuildLogRead, le.Sequence, err)
return nil
}
wake()
return nil
})
if err != nil {
logf("%s is not read (%v): no want names a kept build's parts or its keeping node until the assigner starts again", builds.BuildLogRead, err)
}
}()
}
@@ -0,0 +1,46 @@
package main
import (
"reflect"
"testing"
"time"
"git.novox.be/novox/mesh-delivery/modules/mesh-delivery-builder/builds"
)
// Design 47 G §526, §379, §671: a done build's end in the builder's log, kept by module and fingerprint, names a kept build's
// parts by digest (`<module>:<artifact>`) and its keeping node; a build not kept, failed or not ended names none.
func TestAKeptBuildsEndNamesItsPartsAndItsNode(t *testing.T) {
at := time.Date(2026, 11, 11, 22, 0, 0, 0, time.UTC)
entry := func(event, outcome, keptOn string, made ...builds.BuiltArtifact) []byte {
raw, err := builds.EncodeBuildLogEntry(builds.BuildLogEntry{Event: event, Module: "node-engine", FP: "src2:a", Source: "s", Repository: "novox/mesh-host",
Directory: ".", Commit: "c1", Node: "ace", At: at, Outcome: outcome, Made: made, KeptOn: keptOn})
if err != nil {
t.Fatal(err)
}
return raw
}
made := builds.BuiltArtifact{Name: "program", Kind: "program", Digest: "sha256:1", Reference: "artifact-store://node-engine/program@sha256:1"}
var b builtEnds
for _, raw := range [][]byte{entry(builds.BuildStartedEvent, "", ""), entry(builds.BuildEndedEvent, builds.BuildFailed, "ace")} {
if err := b.putBuilt(raw); err != nil {
t.Fatal(err)
}
}
if _, ok := b.keptBuild("node-engine", "src2:a"); ok {
t.Error("a start and a failed end name a kept build")
}
if err := b.putBuilt(entry(builds.BuildEndedEvent, builds.BuildDone, "", made)); err != nil {
t.Fatal(err)
}
if _, ok := b.keptBuild("node-engine", "src2:a"); ok {
t.Error("a build not kept names a keeping node")
}
if err := b.putBuilt(entry(builds.BuildEndedEvent, builds.BuildDone, "ace", made)); err != nil {
t.Fatal(err)
}
k, ok := b.keptBuild("node-engine", "src2:a")
if !ok || k.KeptOn != "ace" || !reflect.DeepEqual(k.Parts, map[string]string{"node-engine:program": "sha256:1"}) {
t.Errorf("kept build = %+v, %v", k, ok)
}
}
@@ -578,6 +578,9 @@ func follow(ctx context.Context, js jetstream.JetStream, nc process.Subscriber,
}
hearReadsChanged(ctx, js, hub, reads, findings, gate.EventsLive, logger.Printf)
hearRunning(ctx, js, w, wake, findings, logger.Printf)
if e, ok := bus.(emitting); ok && e.Conn != nil {
hearBuildLog(e.Conn, &w.ends, wake, logger.Printf)
}
}
if bus != nil && kvs[buckets.Puts] != nil {
p.acts = newPutActs(bus, wake, logger.Printf)
@@ -3,6 +3,7 @@ package main
import (
"encoding/json"
mcseat "git.novox.be/novox/mesh-contracts/health-and-repair/go/seats/meshconditions"
"git.novox.be/novox/mesh-delivery/modules/mesh-delivery-builder/builds"
"git.novox.be/novox/mesh-sdk/go/views"
"mesh-delivery-assigner/internal/assign"
plc "mesh-delivery-assigner/internal/assignment"
@@ -54,6 +55,9 @@ func TestOpensWhatItsManifestNames(t *testing.T) {
if slices.Contains(readFamilies, r) {
continue // a state kept one bucket per node, read by its prefix (ADR 0593 §6), not one bucket
}
if r == builds.BuildLogRead {
continue // the builder's log, read on the connection (buildlog.go), no bucket
}
reads = append(reads, strings.Replace(r, ".", "_", 1))
}
for _, l := range readFamilies {
@@ -325,6 +325,7 @@ func (p *passer) wants(ctx context.Context) {
held, _ := mwc.HeldKey(name, fp)
grant, _ := grantsstate.PartGrantKey(name, fp)
hold, _ := asks.HoldAskKey(name, fp)
m.KeptBuilds = func(fp string) (assign.KeptBuild, bool) { return p.w.ends.keptBuild(name, fp) }
m.Released = p.answeredWith(hold, "release")
m.Kept = p.answeredWith(hold, "keep")
m.Built, m.Held, m.Granted = p.w.built(name, fp), p.w.has(buckets.Wants, held) && !m.Released, p.w.has(grants, grant)
@@ -41,8 +41,8 @@ type world struct {
changes map[string]uint64
// meta is, by bucket and key, the revision and the time of the value held now, where the watch said them (putEntry).
meta map[string]entryMeta
// trunkBuilt is, by module, its newest done trunk build in the builder's log (putBuilt).
trunkBuilt map[string]builds.BuildLogEntry
// built are the done builds' ends the builder's log says, by module and fingerprint (buildlog.go).
ends builtEnds
// assignments is the seat's state assignments assembled node by node from its keys in either form (mesh-sdk
// mmac.Keys; research 072 row 37): the one reader of its keys in the assigner.
assignmentKeys *mmac.Keys
@@ -74,7 +74,7 @@ type parsed struct {
func newWorld() *world {
return &world{values: map[string]map[string][]byte{}, manifests: map[string]parsed{}, healthy: map[string][]string{}, changes: map[string]uint64{},
meta: map[string]entryMeta{}, trunkBuilt: map[string]builds.BuildLogEntry{}, assignmentKeys: mmac.NewKeys(),
meta: map[string]entryMeta{}, assignmentKeys: mmac.NewKeys(),
sourceGroups: map[uint64]string{}, running: map[string]nodeengine.Running{},
modules: map[string]map[string]meshcatalog.ModulesByModuleValue{}, graphs: map[string]map[string]graph.Node{}}
}
@@ -95,11 +95,39 @@ type Module struct {
// source of a want's Self: every want the rules write or keep is derived from it (selfDerived), so a want written Self while
// the node was placed does not outlive the assignments (issue 1126). Nil: no node does.
Self func(node string) bool
// KeptBuilds answers the kept build of the module at a fingerprint, as the builder's `build.ended` names it (design 47 G §526,
// §671): a want at one names its parts and its keeping node, for a node that states the capability WantParts. Nil: none is read.
KeptBuilds func(fp string) (KeptBuild, bool)
// MachineHeld says whether the node's own network condition is open, which holds the machine as a whole: no want moves to it
// until that clears (design 47 G §706 *M's own network*). Nil: none is held.
MachineHeld func(node string) bool
}
// WantParts is the capability a node-engine of G states (ADR 0614 §4, design 47 G §525): it reads a want's `parts` and `kept-on`.
const WantParts = "want-parts"
// KeptBuild is a kept build as the builder's `build.ended` says it: the digest of each of its parts by `<module>:<artifact>`, and
// the build node that keeps it.
type KeptBuild struct {
Parts map[string]string
KeptOn string
}
// withKept is w naming the kept build at its fingerprint, the parts by digest and the keeping node, only while the node's
// engine states WantParts (design 47 G §525, §526); a want of a build that is not kept, or of a node that does not state it,
// is w.
func (n Nodes) withKept(m Module, node string, w mwc.Want) mwc.Want {
if m.KeptBuilds == nil {
return w
}
k, ok := m.KeptBuilds(w.Fp)
if !ok || !slices.Contains(n.Reads[node], WantParts) {
return w
}
w.Parts, w.KeptOn = k.Parts, k.KeptOn
return w
}
func (m Module) selfOn(node string) bool { return m.Self != nil && m.Self(node) }
func (m Module) composition(node, fp string) (ComposedDigest, bool) {
@@ -514,7 +542,7 @@ func Wants(m Module, n Nodes, wants map[string]mwc.Want, lastHealthy map[string]
switch {
case ok:
d, _ := m.composition(node, target)
next = mwc.Want{Fp: target, Composed: d.Digest}
next = n.withKept(m, node, mwc.Want{Fp: target, Composed: d.Digest})
at := m
if target != m.FP {
// the first want at the older fingerprint the node runs: held back from F by the rule, and judged as a want at it
@@ -1583,3 +1583,28 @@ func TestAProviderRunsHealthyByItsRunning(t *testing.T) {
}
}
}
// Design 47 G §525, §526: a want at a kept build names its parts and the node that keeps it, only for a node whose engine states
// the capability want-parts.
func TestAWantNamesAKeptBuildsPartsAndNodeForANodeThatReadsThem(t *testing.T) {
kept := KeptBuild{Parts: map[string]string{"mail:image": "sha256:1"}, KeptOn: "g14"}
for name, row := range map[string]struct {
reads []string
kept func(string) (KeptBuild, bool)
wanted bool
}{
"states it": {[]string{WantParts}, func(fp string) (KeptBuild, bool) { return kept, fp == fNew }, true},
"does not state": {[]string{"form:oci"}, func(fp string) (KeptBuild, bool) { return kept, fp == fNew }, false},
"build not kept": {[]string{WantParts}, func(string) (KeptBuild, bool) { return KeptBuild{}, false }, false},
"no kept builds": {[]string{WantParts}, nil, false},
"another fingerp": {[]string{WantParts}, func(fp string) (KeptBuild, bool) { return kept, fp == fOld }, false},
} {
m, n := ready()
n.Reads = map[string][]string{"ace": row.reads, "g14": row.reads}
m.KeptBuilds = row.kept
w := Wants(m, n, wantedOld(), nil).Wants["ace"]
if w.Fp != fNew || (w.KeptOn == "g14") != row.wanted || (len(w.Parts) == 1) != row.wanted {
t.Errorf("%s: ace's want = %+v", name, w)
}
}
}
@@ -43,9 +43,10 @@ func TestManifest(t *testing.T) {
slices.Sort(reads)
// ADR 0593 §8: it composes nothing, so it reads no candidate, held or resolved value, compose step or build log; it reads
// every node's published compositions to compare them, and the gate's source and merges, for the release that carries a build (ADR 0601 §1).
// Design 47 G §526, §379: it reads the builder's log for the `build.ended` that names a kept build's parts and keeping node.
// DLV-F2r (ADR 0615 §1, ADR 0631 §1): the indexer's manifests, modules and module graph (whose verb edges a want waits
// on, ADR 0614 §2, §3) are read through its first views and views family, none of the indexer's states by reads.
want := []string{"mesh-delivery-builder.builds", "mesh-delivery-gate.merges", "mesh-delivery-gate.source",
want := []string{"mesh-delivery-builder.build-log", "mesh-delivery-builder.builds", "mesh-delivery-gate.merges", "mesh-delivery-gate.source",
"mesh-delivery-teller.answers", "mesh-delivery-watchman.presence", "mesh-grants.grants", "mesh-host.composed",
"mesh-provisioner.provisions", "node-engine.have"}
if !reflect.DeepEqual(reads, want) {
@@ -55,7 +56,7 @@ func TestManifest(t *testing.T) {
named := []string{buckets.Builds, buckets.Merges, buckets.Source, buckets.Answers, buckets.Presence, buckets.Grants,
buckets.ComposedRead, "mesh-provisioner_provisions", buckets.Have}
slices.Sort(named)
for i, r := range want {
for i, r := range slices.DeleteFunc(slices.Clone(want), func(r string) bool { return r == "mesh-delivery-builder.build-log" }) { // a log, read on the connection
if got := strings.Replace(named[i], "_", ".", 1); got != r {
t.Errorf("the buckets package names %q, the manifest reads %q", got, r)
}
@@ -30,6 +30,7 @@
}
],
"reads": [
"mesh-delivery-builder.build-log",
"mesh-delivery-builder.builds",
"mesh-delivery-gate.merges",
"mesh-delivery-gate.source",