diff --git a/contracts/go/clients/meshwantsclient/placed.go b/contracts/go/clients/meshwantsclient/placed.go new file mode 100644 index 00000000..5ebbf357 --- /dev/null +++ b/contracts/go/clients/meshwantsclient/placed.go @@ -0,0 +1,109 @@ +package meshwantsclient + +import ( + "maps" + "slices" + "sync" + "time" + + seat "git.novox.be/novox/mesh-delivery/contracts/go/seats/meshwants" +) + +// Placed is where the assignment half (the seat mesh-wants' assignment graph revisions; design 47 *The revision view*) +// places each module, and since when: the one reading of "does module stand on node" for every part that takes its view +// of the assignment half (the builder's, the indexer's and the checker's toolchain gate), so no part reads the assigner's +// state for it (design 47 *Each part acts on its own entries*; ADR 0596). Its input is the revisions a part's view holds, +// in the order it holds them (assignmentview's Held): Take takes each whole, from its entries' nodes and modules alone, so +// no entry is asked. A pair stands since the revision that first named it in an unbroken run: in the first revision taken +// since the zero time (the module may have run there before), in a later one since that revision's time (its `at`; the +// zero time when it carries none). A revision not newer than the one taken moves nothing. +// +// Safe for use from a view's goroutine and a reader at once. +type Placed struct { + mu sync.Mutex + since map[string]map[string]time.Time // node -> module -> since when it stands + revision int64 // the revision taken last, 0 before the first +} + +// NewPlaced is a Placed that holds no revision yet. +func NewPlaced() *Placed { return &Placed{since: map[string]map[string]time.Time{}} } + +// Take takes one assignment graph revision in and answers, by node, the modules whose standing on it changed (placed or +// removed), each in name order; nothing for a revision not newer than the one taken. +func (p *Placed) Take(r seat.AssignmentRevision) map[string][]string { + p.mu.Lock() + defer p.mu.Unlock() + if r.Revision <= p.revision { + return nil + } + at := time.Time{} + if p.revision > 0 && r.At != nil { + at = *r.At + } + p.revision = r.Revision + now := map[string]map[string]bool{} + for _, e := range r.Entries { + if now[e.Node] == nil { + now[e.Node] = map[string]bool{} + } + now[e.Node][e.Module] = true + } + changed := map[string][]string{} + next := map[string]map[string]time.Time{} + for node, modules := range now { + next[node] = map[string]time.Time{} + for m := range modules { + if s, ok := p.since[node][m]; ok { + next[node][m] = s + continue + } + next[node][m] = at + changed[node] = append(changed[node], m) + } + } + for node, modules := range p.since { + for m := range modules { + if !now[node][m] { + changed[node] = append(changed[node], m) + } + } + } + for node := range changed { + slices.Sort(changed[node]) + } + p.since = next + return changed +} + +// Revision is the assignment graph revision taken last; 0 before the first. +func (p *Placed) Revision() int64 { + p.mu.Lock() + defer p.mu.Unlock() + return p.revision +} + +// Seeded says the revision taken places some module somewhere. +func (p *Placed) Seeded() bool { + p.mu.Lock() + defer p.mu.Unlock() + return len(p.since) > 0 +} + +// Since says whether module stands on node, and since when (the zero time: since before the first revision taken). +func (p *Placed) Since(node, module string) (time.Time, bool) { + p.mu.Lock() + defer p.mu.Unlock() + s, ok := p.since[node][module] + return s, ok +} + +// ByNode is, by node, the modules placed on it in name order. +func (p *Placed) ByNode() map[string][]string { + p.mu.Lock() + defer p.mu.Unlock() + out := make(map[string][]string, len(p.since)) + for node, mods := range p.since { + out[node] = slices.Sorted(maps.Keys(mods)) + } + return out +} diff --git a/contracts/go/clients/meshwantsclient/wants_test.go b/contracts/go/clients/meshwantsclient/wants_test.go index c4cf388e..490fa692 100644 --- a/contracts/go/clients/meshwantsclient/wants_test.go +++ b/contracts/go/clients/meshwantsclient/wants_test.go @@ -6,7 +6,9 @@ import ( "reflect" "strings" "testing" + "time" + seat "git.novox.be/novox/mesh-delivery/contracts/go/seats/meshwants" "git.novox.be/novox/mesh-sdk/go/keys" ) @@ -160,4 +162,50 @@ func TestWantsAreWhatTheEarlierCodecWrote(t *testing.T) { t.Error("a want key is a held key") } }) + + // Design 47 *Each part acts on its own entries*: the placements a part's toolchain gate counts come from the assignment + // half's revisions, never the assigner's state; a pair stands since the zero time in the first revision taken, since its + // revision's time in a later one, and an older or the same revision moves nothing. + t.Run("placed from the assignment half", func(t *testing.T) { + later := time.Date(2026, 11, 10, 18, 0, 0, 0, time.UTC) + rev := func(n int64, at *time.Time, pairs ...string) seat.AssignmentRevision { + r := seat.AssignmentRevision{Revision: n, Kind: "trunk", At: at} + for _, p := range pairs { + node, module, _ := strings.Cut(p, "/") + r.Entries = append(r.Entries, seat.NodeEntryRef{Node: node, Module: module, Digest: "d-" + p}) + } + return r + } + p := NewPlaced() + if p.Seeded() { + t.Fatal("no revision taken, yet seeded") + } + first := later.Add(-time.Hour) + if got := p.Take(rev(3, &first, "ace/mesh-build-go", "g14/mesh-build-go", "g14/mesh-build-node")); !reflect.DeepEqual(got, + map[string][]string{"ace": {"mesh-build-go"}, "g14": {"mesh-build-go", "mesh-build-node"}}) { + t.Fatalf("first revision changed %v", got) + } + if s, ok := p.Since("ace", "mesh-build-go"); !ok || !s.IsZero() { + t.Fatalf("a pair of the first revision stands since %v (%v), want the zero time", s, ok) + } + if got := p.Take(rev(4, &later, "ace/mesh-build-go", "ace/mesh-build-node", "g14/mesh-build-go")); !reflect.DeepEqual(got, + map[string][]string{"ace": {"mesh-build-node"}, "g14": {"mesh-build-node"}}) { + t.Fatalf("revision 4 changed %v, want mesh-build-node placed on ace and taken off g14", got) + } + if s, ok := p.Since("ace", "mesh-build-node"); !ok || !s.Equal(later) { + t.Fatalf("a pair new in revision 4 stands since %v (%v), want its time %v", s, ok, later) + } + if s, _ := p.Since("ace", "mesh-build-go"); !s.IsZero() { + t.Fatalf("a pair named again keeps its time, got %v", s) + } + if _, ok := p.Since("g14", "mesh-build-node"); ok { + t.Fatal("a pair revision 4 does not name still stands") + } + if got := p.Take(rev(4, &later)); got != nil || !p.Seeded() || p.Revision() != 4 { + t.Fatalf("the same revision again moved %v", got) + } + if got := p.ByNode(); !reflect.DeepEqual(got, map[string][]string{"ace": {"mesh-build-go", "mesh-build-node"}, "g14": {"mesh-build-go"}}) { + t.Fatalf("by node %v", got) + } + }) } diff --git a/modules/mesh-delivery-builder/buildrun/go.mod b/modules/mesh-delivery-builder/buildrun/go.mod index 712cc087..9255332a 100644 --- a/modules/mesh-delivery-builder/buildrun/go.mod +++ b/modules/mesh-delivery-builder/buildrun/go.mod @@ -7,6 +7,8 @@ require ( git.novox.be/novox/mesh-contracts/go/clients/nodeengineclient v0.0.0-00010101000000-000000000000 git.novox.be/novox/mesh-contracts/go/seats/meshmoduleassigner v0.0.0-00010101000000-000000000000 git.novox.be/novox/mesh-contracts/health-and-repair/go/seats/meshwatchdogs v0.0.0-00010101000000-000000000000 + git.novox.be/novox/mesh-delivery/contracts/go/clients/meshwantsclient v0.0.0-00010101000000-000000000000 + git.novox.be/novox/mesh-delivery/contracts/go/seats/meshwants v0.0.0-00010101000000-000000000000 git.novox.be/novox/mesh-delivery/sdk v0.0.0-00010101000000-000000000000 git.novox.be/novox/mesh-sdk/go v0.0.0-00010101000000-000000000000 github.com/nats-io/nats.go v1.54.0 diff --git a/modules/mesh-delivery-builder/buildrun/standing.go b/modules/mesh-delivery-builder/buildrun/standing.go index bc469fc9..f97ae7c0 100644 --- a/modules/mesh-delivery-builder/buildrun/standing.go +++ b/modules/mesh-delivery-builder/buildrun/standing.go @@ -13,6 +13,8 @@ import ( "github.com/nats-io/nats.go/jetstream" "git.novox.be/novox/mesh-contracts/go/clients/meshmoduleassignerclient" + "git.novox.be/novox/mesh-delivery/contracts/go/clients/meshwantsclient" + meshwants "git.novox.be/novox/mesh-delivery/contracts/go/seats/meshwants" "git.novox.be/novox/mesh-delivery/sdk/buildstep" "git.novox.be/novox/mesh-sdk/go/process" ) @@ -40,7 +42,7 @@ type ToolchainGate struct { modules map[string]bool // the build modules whose toolchains pass the gate forward func(module, key string, value []byte, deleted bool) bool // answers whether it changed what it holds logf func(format string, a ...any) - placed *meshmoduleassignerclient.Placed + placed placements known bool // the assignments is settled raw map[string]map[string][]byte // module -> toolchains key -> value, as the bucket holds it left map[string]string // / -> why it was said left out, so each reason is said once @@ -49,6 +51,14 @@ type ToolchainGate struct { OnPlaced func() } +// placements is where the gate reads which modules stand on which node, and since when: the assigner's state +// (meshmoduleassignerclient.Placed, fed by Assignments) or the assignment half (meshwantsclient.Placed, fed by Revision). +type placements interface { + Seeded() bool + Since(node, module string) (time.Time, bool) + ByNode() map[string][]string +} + // NewToolchainGate is a gate over the toolchains of modules, passing what stands on to forward and saying with logf what // it leaves out. func NewToolchainGate(modules []string, forward func(module, key string, value []byte, deleted bool) bool, logf func(string, ...any)) *ToolchainGate { @@ -93,8 +103,13 @@ func (g *ToolchainGate) Toolchain(module, key string, value []byte, deleted bool func (g *ToolchainGate) Assignments(key string, value []byte, deleted bool, at time.Time) bool { g.mu.Lock() defer g.mu.Unlock() - was := g.placed.Seeded() - node, changed, err := g.placed.Put(key, value, deleted, at) + placed, ok := g.placed.(*meshmoduleassignerclient.Placed) + if !ok { + g.logf("%s: %s is not read, the gate takes the assignment half (design 47)", AssignmentsBucket, key) + return false + } + was := placed.Seeded() + node, changed, err := placed.Put(key, value, deleted, at) if err != nil { g.logf("%s: %v (issue 1940)", AssignmentsBucket, err) } @@ -108,6 +123,35 @@ func (g *ToolchainGate) Assignments(key string, value []byte, deleted bool, at t return g.again(node, changed) } +// Revision takes one assignment graph revision the part's view holds in (design 47 *Each part acts on its own entries*: +// the placements come from the seat mesh-wants' assignment half, never the assigner's state; assignmentview's Held hands +// each) and answers whether what it passed on changed. The first revision taken settles the gate, as the assignments' +// history read does; a later one passes on again the reports of each node whose build modules' standing changed. A gate +// fed by Revision reads no Assignments. +func (g *ToolchainGate) Revision(r meshwants.AssignmentRevision) bool { + g.mu.Lock() + defer g.mu.Unlock() + placed, ok := g.placed.(*meshwantsclient.Placed) + if !ok { + placed = meshwantsclient.NewPlaced() + g.placed = placed + } + was := placed.Seeded() + changed := placed.Take(r) + if changed == nil { + return false + } + if !g.known || was != placed.Seeded() { + g.known = true + return g.all() + } + moved := false + for _, node := range slices.Sorted(maps.Keys(changed)) { + moved = g.again(node, changed[node]) || moved + } + return moved +} + // Nodes are the nodes the assignments places any module on, sorted: the nodes the mesh counts (a node leaves the count // when the assignments no longer names it; ADR 0624 ยง1). None until the assignments is read. func (g *ToolchainGate) Nodes() []string { diff --git a/modules/mesh-delivery-builder/buildrun/standing_test.go b/modules/mesh-delivery-builder/buildrun/standing_test.go index cf875aa6..5fca15d6 100644 --- a/modules/mesh-delivery-builder/buildrun/standing_test.go +++ b/modules/mesh-delivery-builder/buildrun/standing_test.go @@ -14,6 +14,7 @@ import ( "git.novox.be/novox/mesh-contracts/go/clients/meshmoduleassignerclient" "git.novox.be/novox/mesh-contracts/go/seats/meshmoduleassigner" + meshwants "git.novox.be/novox/mesh-delivery/contracts/go/seats/meshwants" "git.novox.be/novox/mesh-delivery/sdk/buildstep" ) @@ -188,6 +189,48 @@ func TestToolchainGate(t *testing.T) { } }) + // Design 47 *Each part acts on its own entries*: fed by the assignment half's revisions, the gate settles at the first, + // leaves out a node no revision places the build module on, and counts a node placed in a later revision only by a + // report written since that revision's time. + t.Run("fed by the assignment half", func(t *testing.T) { + set := &ToolchainSet{} + said := &[]string{} + gate := NewToolchainGate([]string{GoBuildModule, NodeBuildModule}, set.Put, func(f string, a ...any) { *said = append(*said, fmt.Sprintf(f, a...)) }) + for _, node := range append([]string{"ace"}, three...) { + key, value := gateReport(t, node, "v26.12.1", placed.Add(time.Minute)) + if node == "ace" { + key, value = gateReport(t, node, "v26.12.0", old) + } + gate.Toolchain(NodeBuildModule, key, value, false) + } + revision := func(n int64, at time.Time, nodes ...string) meshwants.AssignmentRevision { + r := meshwants.AssignmentRevision{Revision: n, Kind: "trunk", At: &at} + for _, node := range nodes { + r.Entries = append(r.Entries, meshwants.NodeEntryRef{Node: node, Module: NodeBuildModule, Digest: "d-" + node}) + } + return r + } + if !gate.Revision(revision(7, old.Add(-time.Hour), three...)) || len(set.Entries("typescript")) != 3 { + t.Fatalf("the first revision settles the gate and leaves ace out; %d entries", len(set.Entries("typescript"))) + } + if gate.Revision(revision(7, old.Add(-time.Hour), three...)) { + t.Fatal("the same revision again changed what was passed on") + } + if gate.Revision(revision(8, placed, "ace", "g14", "novox", "shanks")); len(set.Entries("typescript")) != 3 { + t.Fatal("ace's report from before revision 8 placed it counts") + } + if len(*said) < 2 || !strings.Contains((*said)[len(*said)-1], "before mesh-build-node was placed on ace at 2026-10-25T19:56:45Z") { + t.Fatalf("said %q, want why ace's old report is left out", *said) + } + key, value := gateReport(t, "ace", "v26.12.1", placed.Add(time.Minute)) + if !gate.Toolchain(NodeBuildModule, key, value, false) || len(set.Entries("typescript")) != 4 { + t.Fatal("ace's report written since revision 8 is not passed on") + } + if !gate.Revision(revision(9, placed.Add(time.Hour), "ace", "novox", "shanks")) || len(set.Entries("typescript")) != 3 || len(gate.Nodes()) != 3 { + t.Fatalf("g14 taken off in revision 9: %d entries, nodes %v", len(set.Entries("typescript")), gate.Nodes()) + } + }) + t.Run("the history: a module in the first revision heard of a key stands since before it", func(t *testing.T) { p := meshmoduleassignerclient.NewPlaced() // ace's node key is the first revision heard of it: its module stands since before.