B2l AS17 (4): §298's three order edges are acting waits of the assigner under order.

The new node-backup and bus module holders run before a module whose kept data's form changes moves; at a revert the node-engine goes
back last on its node; the reverted assigner and gate are wanted only once take-in.taken names the trunk's head (A-260). Kept in the
state waits, said as wait-held and wait-ended, read and released by assigner_waits and assigner_release. The manifest consumes the
indexer's take-in.taken and its version is 0.3 (A-245).

Builds: design 47 G §298, A-245, A-258, A-259, A-260
Claude-Session: https://claude.ai/code/session_01NcbgfGKaxw984XyAPfLm4h
This commit is contained in:
2026-11-12 09:49:25 +01:00
parent 1e48d04077
commit cd51fec81a
13 changed files with 740 additions and 19 deletions
@@ -55,7 +55,7 @@ func (p *passer) entryEdges(node, module string) []seat.Edge {
func (p *passer) actingReleased() map[string]bool {
out := map[string]bool{}
for key, raw := range p.w.read(buckets.Waits) {
if w, err := acting.Decode(raw); err == nil && w.Released && acting.Is(key) {
if w, err := acting.Decode(raw); err == nil && w.Released && (acting.Is(key) || acting.IsOrder(key)) {
out[key+"@"+w.Fp] = true
}
}
@@ -105,7 +105,7 @@ func (r *releasedVersions) of(module string, source int64) (assign.KeptBuild, bo
}
// hearBuildsAndReleases registers the hub's takers of build.ended and version-released, each waking the pass; before the hub starts.
func hearBuildsAndReleases(hub *process.EventHub, ends *builtEnds, released *releasedVersions, wake func(), logf func(string, ...any)) error {
func hearBuildsAndReleases(hub *process.EventHub, ends *builtEnds, released *releasedVersions, taken *takenSources, wake func(), logf func(string, ...any)) error {
taker := func(put func([]byte) error) func(jetstream.Msg) {
return func(m jetstream.Msg) {
if err := put(m.Data()); err != nil {
@@ -120,5 +120,8 @@ func hearBuildsAndReleases(hub *process.EventHub, ends *builtEnds, released *rel
if err := hub.Handle(buildEndedSubjects, taker(ends.put)); err != nil {
return err
}
return hub.Handle(meshcatalog.SubjectVersionReleased, taker(released.put))
if err := hub.Handle(meshcatalog.SubjectVersionReleased, taker(released.put)); err != nil {
return err
}
return hub.Handle(takenSubject, taker(taken.put))
}
@@ -604,7 +604,7 @@ func follow(ctx context.Context, js jetstream.JetStream, nc process.Subscriber,
logger.Printf("a pipeline's end or a registration is not heard, so the wants keep the trunk-only form until the assigner starts again: %v", err)
}
}
if err := hearBuildsAndReleases(hub, &w.ends, &w.released, wake, logger.Printf); err != nil {
if err := hearBuildsAndReleases(hub, &w.ends, &w.released, &w.taken, wake, logger.Printf); err != nil {
logger.Printf("a build's end or a version's release is not heard, so no want names a build's parts, its keeping node or its version until the assigner starts again: %v", err)
}
hearReadsChanged(ctx, js, hub, reads, findings, gate.EventsLive, logger.Printf)
@@ -0,0 +1,296 @@
package main
import (
"context"
"encoding/json"
"fmt"
"strconv"
"sync"
"git.novox.be/novox/mesh-contracts/go/seats/meshbroker"
mma "git.novox.be/novox/mesh-contracts/go/seats/meshmoduleassigner"
"git.novox.be/novox/mesh-contracts/go/seats/nodeengine"
mwc "git.novox.be/novox/mesh-delivery/modules/mesh-delivery-contracts/go/clients/meshwantsclient"
"git.novox.be/novox/mesh-delivery/modules/mesh-delivery-contracts/go/seats/meshcatalog"
"git.novox.be/novox/mesh-delivery/modules/mesh-delivery-contracts/go/seats/meshdelivery"
seat "git.novox.be/novox/mesh-delivery/modules/mesh-delivery-contracts/go/seats/meshwants"
"git.novox.be/novox/mesh-sdk/go/catalogue"
"git.novox.be/novox/mesh-sdk/go/process"
"git.novox.be/novox/mesh-sdk/go/seatdef"
"mesh-delivery-assigner/internal/acting"
"mesh-delivery-assigner/internal/assign"
"mesh-delivery-assigner/internal/assigngraph"
"mesh-delivery-assigner/internal/assignment"
"mesh-delivery-assigner/internal/buckets"
)
// The three order edges of a delivery that crosses a form change (design 47 G §298, A-245, A-258, A-259, A-260). Each is an acting wait of the
// assigner, kept in the state `waits` under `order.<node>.<module>.<part>` beside the acting waits, in the step placing, said as wait-held
// when a want is held for it and wait-ended (ran, released, replaced) when it moves; assigner_waits reads them and assigner_release with
// how `now` releases them, as it does an acting wait. The holders of the seats are read from the seat holders the assignments compose
// (assignment.SeatHoldersOf), never named:
// - the node-backup holder's and the bus module's (the holder of the seat mesh-broker) new versions run before a module whose kept data's
// form changes is moved, since the restore point needs both (OrderBackup);
// - at a revert a node's node-engine goes back last on that node, after every module it restores there runs the reverted build, which it
// starts only once it has said `restored` (OrderEngineLast);
// - the reverted assigner and gate are wanted only once the indexer's take-in.taken of the trunk branch names a source revision at least
// the trunk's head, the highest `src` of the trunk branch's module values (OrderTakenIn, A-260).
// takenSubject is the indexer's event take-in.taken on its own stream, as the manifest's consumes names it.
var takenSubject = process.ModuleStreamSubject(buckets.IndexerModule, "take-in.taken")
// takenSources are the newest source revisions a take-in.taken named, by watched branch, as heard (hearBuildsAndReleases).
type takenSources struct {
mu sync.Mutex
own map[string]int64
}
// put takes one take-in.taken in. A hand drop, and a taken with no source revision (one said before G), take nothing in.
func (t *takenSources) put(data []byte) error {
var e meshcatalog.TakeInTaken
if err := json.Unmarshal(data, &e); err != nil {
return fmt.Errorf("a take-in.taken that is none: %w", err)
}
if e.Dropped || e.Source == 0 || e.Branch == "" {
return nil
}
t.mu.Lock()
defer t.mu.Unlock()
if t.own == nil {
t.own = map[string]int64{}
}
t.own[e.Branch] = max(t.own[e.Branch], e.Source)
return nil
}
// of is the newest source revision of branch taken in; 0 before one is heard.
func (t *takenSources) of(branch string) int64 {
t.mu.Lock()
defer t.mu.Unlock()
return t.own[branch]
}
// orderEdge is one order edge held on module's want on node: part's version fp must run on partNode (OrderBackup, OrderEngineLast), or the
// trunk's source revision source be taken in (OrderTakenIn).
type orderEdge struct {
node, module, part, partNode, fp, entry, order string
source int64
}
func (e orderEdge) key() string { return acting.OrderKey(e.node, e.module, e.part) }
func (e orderEdge) acting() assign.ActingEdge {
return assign.ActingEdge{Node: e.partNode, Part: e.part, Fp: e.fp, Step: seatdef.StepPlace}
}
// orderRuns says an order edge is met: its part runs the version, or the source revision is taken in.
func (p *passer) orderRuns(n assign.Nodes, e orderEdge) bool {
if e.order == acting.OrderTakenIn {
return p.w.taken.of(trunkBranch) >= e.source
}
return n.Runs(e.acting())
}
// seatHolders are the holders of each seat as the assignments compose them, by seat name.
func (p *passer) holdersBySeat() map[string][]mma.SeatHoldersHoldersItem {
out := map[string][]mma.SeatHoldersHoldersItem{}
for _, sh := range assignment.SeatHoldersOf(p.w.assignments()) {
out[sh.Seat] = append(out[sh.Seat], sh.Holders...)
}
return out
}
// orderEdgesOf computes §298's edges on the wants of module m on each of its nodes, into p.orders. wanted is m's wants by node; trunk the
// trunk's module values. A node that has no running build heard, runs m at its trunk version already, or is wanted that version already
// has none: nothing moves. It answers, by node, whether an edge still stands that cannot be recorded as a wait because the catalog entry
// it reads is not answered yet (asked, and its answer wakes a pass): the want is held meanwhile, never written across a form change unread.
func (p *passer) orderEdgesOf(ctx context.Context, m assign.Module, wanted map[string]mwc.Want, trunk map[string]meshcatalog.ModulesByModuleValue) map[string]bool {
unread := map[string]bool{}
p.orderVisited[m.Name] = true
if m.FP == "" || p.graph == nil || p.entries == nil {
return unread
}
var holders map[string][]mma.SeatHoldersHoldersItem
for _, node := range m.Nodes {
if w, ok := wanted[node]; ok && !w.Removed && w.Fp == m.FP {
continue // the version is wanted already: nothing moves, so nothing waits
}
run, heard := p.w.runningOf(node, m.Name)
if !heard || run.Stopped || run.Fingerprint == "" || run.Fingerprint == m.FP {
continue
}
if holders == nil {
holders = p.holdersBySeat()
}
held := func(seatName string) bool {
for _, h := range holders[seatName] {
if h.Module == m.Name && h.Node == node {
return true
}
}
return false
}
add := func(e orderEdge) {
e.node, e.module, e.entry = node, m.Name, m.FP
p.orders[e.key()] = e
}
// 1. a kept data's form changes: the restore point needs the new node-backup holder and bus module.
changes, known := p.formChange(ctx, m.Name, run.Fingerprint, m.FP)
if !known {
unread[node] = true
}
if len(changes) > 0 {
for seatName, onNode := range map[string]bool{catalogue.BackupSeat: true, meshbroker.Seat: false} {
for _, h := range holders[seatName] {
if h.Module == m.Name || (onNode && h.Node != node) || trunk[h.Module].Fp2 == "" {
continue
}
add(orderEdge{part: h.Module, partNode: h.Node, fp: trunk[h.Module].Fp2, order: acting.OrderBackup})
}
}
}
if !held(nodeengine.Seat) && !held(mma.Seat) && !held(meshdelivery.Seat) {
continue
}
reverted, err := assigngraph.Reverted(ctx, p.graph.objects, node, m.Name, run.Fingerprint, m.FP)
if err != nil {
p.logf("whether the want of %s on %s at %s is a revert is not read: %v", m.Name, node, m.FP, err)
continue
}
if !reverted {
continue
}
// 2. at a revert the node-engine goes back last on its node: after every module that is set back there across a form change runs.
if held(nodeengine.Seat) {
for module, byNode := range p.w.wantsOf() {
// What it is set back to is its trunk version, which the rules write as its want in this pass.
target := trunk[module].Fp2
if w, ok := byNode[node]; module == m.Name || !ok || w.Removed || target == "" {
continue
}
xrun, heard := p.w.runningOf(node, module)
if !heard || xrun.Stopped || xrun.Fingerprint == "" || xrun.Fingerprint == target {
continue
}
if back, err := assigngraph.Reverted(ctx, p.graph.objects, node, module, xrun.Fingerprint, target); err != nil || !back {
continue
}
xchanges, known := p.formChange(ctx, module, xrun.Fingerprint, target)
if !known {
unread[node] = true
}
if len(xchanges) > 0 {
add(orderEdge{part: module, partNode: node, fp: target, order: acting.OrderEngineLast})
}
}
}
// 3. the reverted assigner and gate wait for the indexer to take in the trunk's head.
if held(mma.Seat) || held(meshdelivery.Seat) {
var head int64
for _, v := range p.w.trunkOf(trunkBranch) {
head = max(head, v.Src)
}
if head > 0 {
add(orderEdge{part: buckets.IndexerModule, fp: strconv.FormatInt(head, 10), source: head, order: acting.OrderTakenIn})
}
}
}
return unread
}
// orderActing is the order edges of module on node that still hold its want, as acting edges: those not met, and not released by hand. A
// node whose catalog entry is not read yet holds its want too, on a stand-in edge no part runs.
func (p *passer) orderActing(module string, released map[string]bool, unread map[string]bool, n func() assign.Nodes) func(node string) []assign.ActingEdge {
return func(node string) []assign.ActingEdge {
var out []assign.ActingEdge
nodes := n()
for _, e := range p.orders {
if e.module == module && e.node == node && !released[e.key()+"@"+e.fp] && !p.orderRuns(nodes, e) {
out = append(out, assign.ActingEdge{Node: e.partNode, Part: e.part, Fp: e.fp, Step: seatdef.StepPlace})
}
}
if unread[node] {
out = append(out, assign.ActingEdge{Node: node, Part: "the catalog entry", Step: seatdef.StepPlace})
}
return out
}
}
// orderWaits keeps the order waits after the pass's wants, as actingWaits keeps the acting waits: it starts the wait of each edge not met,
// ends the wait of one that ran, was replaced or was released, and says each.
func (p *passer) orderWaits(ctx context.Context, n assign.Nodes, wanted map[string]map[string]mwc.Want) {
if p.kvs[buckets.Waits] == nil {
return
}
now := p.clock()
kept := p.w.read(buckets.Waits)
trunk := p.w.trunk()
if p.orderSaid == nil {
p.orderSaid = map[string]bool{}
}
standing := map[string]bool{}
for key, e := range p.orders {
if w, ok := wanted[e.module][e.node]; ok && !w.Removed && w.Fp == e.entry {
continue
}
if p.orderRuns(n, e) {
continue
}
standing[key] = true
if raw := kept[key]; raw != nil {
if w, err := acting.Decode(raw); err == nil && w.Entry == e.entry && w.Fp == e.fp {
if !w.Released && !p.orderSaid[key] {
// A wait still held after a restart is said again, as wait-held (A-260).
if err := p.wantsSay(seat.EventWaitHeld, acting.Body(w, assignment.Module), key+"."+e.fp+"."+e.entry+".again"); err != nil {
p.logf("the wait of %s on %s for %s is not said again, and is said at the next pass: %v", e.module, e.node, e.part, err)
} else {
p.orderSaid[key] = true
}
}
p.actingReleaseSaid(ctx, key, w)
continue
}
}
if !n.Online[e.node] {
continue // a node away holds only itself, and its clock has not started (§268)
}
w := acting.Wait{Step: seatdef.StepPlace, Node: e.node, Module: e.module, Part: e.part, PartNode: e.partNode, Fp: e.fp, Entry: e.entry, Order: e.order,
Since: now, ClockFrom: now, Bound: acting.BoundOf(seat.Edge{Step: seatdef.StepPlace}).String()}
if err := p.wantsSay(seat.EventWaitHeld, acting.Body(w, assignment.Module), key+"."+e.fp+"."+e.entry); err != nil {
p.logf("the wait of %s on %s for %s is not said, and is said at the next pass: %v", e.module, e.node, e.part, err)
continue
}
p.put(ctx, buckets.Waits, key, func() ([]byte, error) { return acting.Encode(w) })
p.orderSaid[key] = true
}
for key, raw := range kept {
if !acting.IsOrder(key) || standing[key] {
continue
}
w, err := acting.Decode(raw)
if err != nil {
continue
}
if _, there := trunk[w.Module]; there && !p.orderVisited[w.Module] {
continue // the module was not looked at in this pass (no pipeline passed for it yet): its wait stands as kept
}
how := acting.HowRan
switch e, there := p.orders[key]; {
case w.Released:
how = acting.HowReleased
case there && e.fp != w.Fp, !there && wanted[w.Module][w.Node].Fp != w.Entry:
how = acting.HowReplaced
}
if !w.EndSaid {
body := acting.Body(w, assignment.Module)
body.How = how
if err := p.wantsSay(seat.EventWaitEnded, body, key+"."+w.Fp+"."+w.Entry+"."+how); err != nil {
p.logf("the end of the wait of %s on %s for %s is not said, and is said at the next pass: %v", w.Module, w.Node, w.Part, err)
continue
}
}
p.remove(ctx, buckets.Waits, key)
delete(p.orderSaid, key)
}
}
@@ -0,0 +1,355 @@
package main
import (
"context"
"encoding/json"
"fmt"
"strings"
"testing"
"time"
mmac "git.novox.be/novox/mesh-contracts/go/clients/meshmoduleassignerclient"
"git.novox.be/novox/mesh-contracts/go/clients/nodeengineclient"
"git.novox.be/novox/mesh-contracts/go/seats/meshbroker"
mma "git.novox.be/novox/mesh-contracts/go/seats/meshmoduleassigner"
"git.novox.be/novox/mesh-contracts/go/seats/nodeengine"
oc "git.novox.be/novox/mesh-contracts/operator-and-conversation/go/seats/operatorchannel"
"git.novox.be/novox/mesh-delivery/modules/mesh-delivery-builder/builds"
"git.novox.be/novox/mesh-delivery/modules/mesh-delivery-contracts/go/clients/meshcatalogclient"
mwc "git.novox.be/novox/mesh-delivery/modules/mesh-delivery-contracts/go/clients/meshwantsclient"
"git.novox.be/novox/mesh-delivery/modules/mesh-delivery-contracts/go/seats/meshcatalog"
"git.novox.be/novox/mesh-delivery/modules/mesh-delivery-contracts/go/seats/meshdelivery"
meshwants "git.novox.be/novox/mesh-delivery/modules/mesh-delivery-contracts/go/seats/meshwants"
"git.novox.be/novox/mesh-delivery/modules/mesh-grants/grantsstate"
"git.novox.be/novox/mesh-sdk/go/catalogue"
"git.novox.be/novox/mesh-sdk/go/seatdef"
"mesh-delivery-assigner/internal/acting"
"mesh-delivery-assigner/internal/assigngraph"
plc "mesh-delivery-assigner/internal/assignment"
"mesh-delivery-assigner/internal/buckets"
)
// orderFP is a fingerprint of the form the trunk keeps, made of one letter.
func orderFP(c string) string {
return meshcatalogclient.SourcePrefix2 + strings.Repeat(fmt.Sprintf("%x", c[0]&15), 64)
}
// orderMod is one module of an order test: the fingerprint it runs on ace and the one the trunk holds, the form of its kept data at each
// fingerprint (0: it keeps none), the seats it holds and the source revision of its trunk value.
type orderMod struct {
name string
run, trunk string
forms map[string]int
holds []string
src int64
}
type orderFixture struct {
t *testing.T
p *passer
wants *memKV
events []string
bodies []meshwants.Wait
answers chan struct{}
}
// newOrderFixture is a pass on node ace, which composes its own modules, with each module running its fingerprint healthy, built and
// granted at the trunk's, an older want, the catalog's entries answered and each held seat as the assignments say it. history lists the
// fingerprints of each module by assignment graph revision, oldest first (the objects Reverted reads).
func newOrderFixture(t *testing.T, mods []orderMod, history ...map[string]string) *orderFixture {
t.Helper()
ctx := context.Background()
f := &orderFixture{t: t, wants: &memKV{values: map[string][]byte{}}, answers: make(chan struct{}, 64)}
f.p = newPasser(kvSet{buckets.Wants: f.wants, buckets.Asks: &memKV{values: map[string][]byte{}}, buckets.Waits: &memKV{values: map[string][]byte{}}}, time.Now())
p := f.p
p.logf = t.Logf
p.say = func(oc.Condition) error { return nil }
p.publishWants = func(event string, body any, id string) error {
f.events = append(f.events, event)
f.bodies = append(f.bodies, body.(meshwants.Wait))
return nil
}
digest := "sha256:" + strings.Repeat("c", 64)
pl := mma.Assignments{Node: "ace"}
forms := map[string]int{} // by fingerprint: each module's own
modules := []string{}
for _, m := range mods {
modules = append(modules, m.name)
pl.Modules = append(pl.Modules, mma.PlacedModule{Module: m.name})
for _, seat := range m.holds {
scope := "node"
if seat != catalogue.BackupSeat && seat != nodeengine.Seat {
scope = "mesh"
}
pl.Holds = append(pl.Holds, mma.Holding{Claim: seat, Module: m.name, Scope: scope})
}
for fp, form := range m.forms {
forms[fp] = form
}
mdigest := fmt.Sprintf("%064x", len(modules))
mod, _ := meshcatalogclient.EncodeModule(meshcatalog.ModulesByModuleValue{Fp2: m.trunk, Manifest: mdigest, Repository: "novox/" + m.name, Commit: strings.Repeat("1", 40), Path: meshcatalogclient.RootPath, Src: m.src})
putModuleRaw(p.w, trunkBranch, m.name, mod)
p.w.manifests[mdigest] = parsed{module: m.name, manifest: catalogue.Manifest{Module: m.name}}
bkey, _ := builds.BuilderBuildKey(m.name, m.trunk)
built, _ := builds.EncodeBuilderBuild(builds.BuilderBuild{State: builds.BuildDone, Manifest: mdigest})
p.w.put(buckets.Builds, bkey, built)
gkey, _ := grantsstate.PartGrantKey(m.name, m.trunk)
grant, _ := grantsstate.EncodeGrant(grantsstate.Grant{})
p.w.put(buckets.Grants, gkey, grant)
f.runs(m.name, m.run)
wkey, _ := mwc.WantKey("ace", m.name)
old, _ := mwc.EncodeWant(mwc.Want{Fp: m.run, Self: true})
p.w.put(buckets.Wants, wkey, old)
f.wants.values[wkey] = old
}
composes(t, p.w, "ace", digest)
values, err := mmac.AssignmentsValues(pl)
if err != nil {
t.Fatal(err)
}
for key, raw := range values {
p.w.put(plc.AssignmentsBucket, key, raw)
}
on, _ := presenceValue(true)
p.w.put(buckets.Presence, "ace", on)
objects := memObjects{}
for i, at := range history {
half := meshcatalogclient.Half{Revision: meshcatalog.CatalogRevision{Branch: trunkBranch, Kind: "trunk", Revision: int64(i + 1), Source: int64(i + 1)},
Entries: map[string]meshcatalog.CatalogEntry{}}
for _, module := range modules {
digest := fmt.Sprintf("sha256:%064x", (i+1)*1000+len(half.Entries))
half.Revision.Entries = append(half.Revision.Entries, meshcatalog.EntryRef{Module: module, Digest: digest})
half.Entries[module] = meshcatalog.CatalogEntry{Module: module, Fp: at[module]}
}
placed := mma.Assignments{Node: "ace"}
for _, module := range modules {
placed.Modules = append(placed.Modules, mma.PlacedModule{Module: module})
}
in := assigngraph.Inputs{Halves: map[string]meshcatalogclient.Half{trunkBranch: half}, FirstBranch: func(string) string { return trunkBranch },
Placements: map[string]mma.Assignments{"ace": placed}}
if _, _, err := assigngraph.Keep(ctx, objects, assigngraph.Compute(in), "test", time.Now()); err != nil {
t.Fatal(err)
}
}
p.graph = &graphKeeper{objects: objects}
p.entries = meshcatalogclient.NewEntries(func(_ string, body any) (json.RawMessage, error) {
in := body.(meshcatalog.EntryInput)
entry := meshcatalog.CatalogEntry{Module: "m", Fp: in.Fingerprint, Manifest: map[string]any{"module": "m"}}
if form := forms[in.Fingerprint]; form > 0 {
entry.Manifest = map[string]any{"module": "m", "state": []any{map[string]any{"name": "sessions", "form": form}}}
}
return json.Marshal(meshcatalog.EntryAnswer{Entry: entry})
})
p.entries.Changed = func(meshcatalogclient.EntryRow) { f.answers <- struct{}{} }
return f
}
// runs says module runs fp healthy on ace, heard as the node-engine seat's running and in the have.
func (f *orderFixture) runs(module, fp string) {
have, _ := nodeengineclient.EncodeHave(nodeengine.HaveByNodeModuleValue{Fp: fp, Composed: "sha256:" + strings.Repeat("c", 64), Phase: nodeengineclient.PhaseRunning, Health: nodeengineclient.HealthOK})
f.p.w.put(buckets.Have, "ace."+module, have)
f.p.w.running["ace."+module] = nodeengine.Running{Node: "ace", Module: module, Fingerprint: fp}
}
// pass makes a pass, and after it every catalog entry it asked is answered.
func (f *orderFixture) pass() {
f.t.Helper()
f.p.wants(context.Background())
asked := 0
for _, row := range f.p.entries.Rows() {
if row.State == meshcatalogclient.EntryAsked {
asked++
}
}
for range asked {
select {
case <-f.answers:
case <-time.After(10 * time.Second): // a bound on a failure only: the answers come at once
f.t.Fatal("an entry asked was not answered")
}
}
}
// wantAt is the fingerprint module is wanted at on ace.
func (f *orderFixture) wantAt(module string) string {
f.t.Helper()
key, _ := mwc.WantKey("ace", module)
w, err := mwc.DecodeWant(f.wants.values[key])
if err != nil {
f.t.Fatal(err)
}
return w.Fp
}
func (f *orderFixture) waits() []string {
var out []string
for key := range f.p.w.read(buckets.Waits) {
if acting.IsOrder(key) {
out = append(out, key)
}
}
return out
}
// Design 47 G §298 first edge, A-258, A-259: a module whose kept data's form changes is not moved until the new version of the node-backup
// holder and of the holder of the seat mesh-broker run; the wait is kept under `order.` in the state waits, said as wait-held, and ends
// ran (wait-ended) when the holders run; a module whose kept data keeps its form is moved at once.
func TestAModuleWhoseKeptDataChangesFormWaitsForTheNewBackupHolderAndTheBusModule(t *testing.T) {
f := newOrderFixture(t, []orderMod{
{name: "web", run: orderFP("a"), trunk: orderFP("b"), forms: map[string]int{orderFP("a"): 1, orderFP("b"): 2}},
{name: "plain", run: orderFP("c"), trunk: orderFP("d"), forms: map[string]int{orderFP("c"): 1, orderFP("d"): 1}},
{name: "backup", run: orderFP("e"), trunk: orderFP("f"), holds: []string{catalogue.BackupSeat}},
{name: "bus", run: orderFP("g"), trunk: orderFP("h"), holds: []string{meshbroker.Seat}},
})
f.pass() // the entries are asked: no want crosses a form change unread
if got := f.wantAt("web"); got != orderFP("a") {
t.Fatalf("web is wanted at %s before the entries are read", got)
}
f.pass()
if got := f.wantAt("web"); got != orderFP("a") {
t.Fatalf("web moved across a form change at %s while the new node-backup holder and bus module do not run", got)
}
if got := f.wantAt("plain"); got != orderFP("d") {
t.Errorf("a module whose kept data keeps its form is wanted at %s, not moved", got)
}
if len(f.waits()) != 2 || len(f.events) != 2 || f.events[0] != meshwants.EventWaitHeld || f.bodies[0].Step != seatdef.StepPlace || f.bodies[0].Module != "web" {
t.Fatalf("waits %v events %v bodies %+v: want two wait-held in the step place, `order.ace.web.backup` and `order.ace.web.bus`", f.waits(), f.events, f.bodies)
}
if _, err := acting.Decode(f.p.w.read(buckets.Waits)[acting.OrderKey("ace", "web", "backup")]); err != nil {
t.Fatalf("the wait is not kept under its key: %v", err)
}
f.pass()
if len(f.events) != 2 {
t.Fatalf("a pass that changed nothing said %v", f.events)
}
f.runs("backup", orderFP("f"))
f.pass()
if got := f.wantAt("web"); got != orderFP("a") || len(f.events) != 3 || f.events[2] != meshwants.EventWaitEnded || f.bodies[2].How != acting.HowRan {
t.Fatalf("with the bus module still old: web at %s, events %v", got, f.events)
}
f.runs("bus", orderFP("h"))
f.pass()
f.pass()
if got := f.wantAt("web"); got != orderFP("b") || len(f.waits()) != 0 || len(f.events) != 4 {
t.Fatalf("with both running: web at %s, waits %v, events %v", got, f.waits(), f.events)
}
}
// §298 first edge, the hand: assigner_waits reads an order wait and assigner_release with how `now` lets the module move.
func TestAnOrderWaitIsReadAndReleasedByTheExistingTools(t *testing.T) {
f := newOrderFixture(t, []orderMod{
{name: "web", run: orderFP("a"), trunk: orderFP("b"), forms: map[string]int{orderFP("a"): 1, orderFP("b"): 2}},
{name: "backup", run: orderFP("e"), trunk: orderFP("f"), holds: []string{catalogue.BackupSeat}},
})
f.pass()
f.pass()
var acts []string
s := newPutTools(memStore{waitsState: f.p.kvs[buckets.Waits].(*memKV)}, func(key string, body any) (json.RawMessage, error) {
acts = append(acts, key)
return json.RawMessage(`{"ok":true,"log":"logged"}`), nil
})
got, err := s.doWaits(map[string]any{"module": "web"})
if list := got.(map[string]any)["waits"].([]any); err != nil || len(list) != 1 || list[0].(actingSaid).Key != "order.ace.web.backup" {
t.Fatalf("assigner_waits = %+v %v", got, err)
}
if _, err := s.doRelease(map[string]any{"node": "ace", "module": "web", "how": "now", "why": "the restore point is taken by hand"}, "shanks.node-tools"); err != nil || len(acts) == 0 {
t.Fatalf("%v %v", err, acts)
}
f.p.w.put(buckets.Waits, "order.ace.web.backup", f.p.kvs[buckets.Waits].(*memKV).values["order.ace.web.backup"])
f.pass()
f.pass()
if got := f.wantAt("web"); got != orderFP("b") || f.bodies[len(f.bodies)-1].How != acting.HowReleased {
t.Fatalf("after the release: web at %s, last %+v", got, f.bodies[len(f.bodies)-1])
}
}
// §298 second edge, A-259: at a revert the node-engine goes back last on its node, after every module that is set back there across a
// form change runs the reverted build.
func TestAtARevertTheNodeEngineGoesBackLast(t *testing.T) {
old, now := map[string]string{"web": orderFP("a"), "engine": orderFP("c")}, map[string]string{"web": orderFP("b"), "engine": orderFP("d")}
f := newOrderFixture(t, []orderMod{
{name: "web", run: orderFP("b"), trunk: orderFP("a"), forms: map[string]int{orderFP("a"): 1, orderFP("b"): 2}},
{name: "engine", run: orderFP("d"), trunk: orderFP("c"), holds: []string{nodeengine.Seat}},
}, old, now)
f.pass()
f.pass()
if f.wantAt("web") != orderFP("a") {
t.Fatalf("web is wanted at %s: the module set back is not held", f.wantAt("web"))
}
if got := f.wantAt("engine"); got != orderFP("d") {
t.Fatalf("the node-engine is set back to %s before web runs its reverted build", got)
}
if w := f.waits(); len(w) != 1 || w[0] != "order.ace.engine.web" || f.bodies[len(f.bodies)-1].Module != "engine" {
t.Fatalf("waits %v bodies %+v", w, f.bodies)
}
f.runs("web", orderFP("a"))
f.pass()
f.pass()
if got := f.wantAt("engine"); got != orderFP("c") || len(f.waits()) != 0 {
t.Fatalf("with web back: the node-engine is wanted at %s, waits %v", got, f.waits())
}
}
// §298 third edge, A-259, A-260: the reverted assigner and gate are wanted only once the indexer's take-in.taken of the trunk branch names a
// source revision at least the trunk's head (the highest src of the trunk's module values).
func TestTheRevertedAssignerAndGateWaitForTheIndexerToTakeInTheTrunksHead(t *testing.T) {
old, now := map[string]string{"asg": orderFP("a"), "gate": orderFP("c"), "idx": orderFP("e")}, map[string]string{"asg": orderFP("b"), "gate": orderFP("d"), "idx": orderFP("f")}
f := newOrderFixture(t, []orderMod{
{name: "asg", run: orderFP("b"), trunk: orderFP("a"), holds: []string{mma.Seat}, src: 6},
{name: "gate", run: orderFP("d"), trunk: orderFP("c"), holds: []string{meshdelivery.Seat}, src: 7},
{name: "idx", run: orderFP("e"), trunk: orderFP("e"), src: 7},
}, old, now)
f.pass()
f.pass()
if f.wantAt("asg") != orderFP("b") || f.wantAt("gate") != orderFP("d") {
t.Fatalf("the reverted assigner (%s) and gate (%s) are wanted before the head is taken in", f.wantAt("asg"), f.wantAt("gate"))
}
if len(f.waits()) != 2 || f.bodies[0].Part != "mesh-delivery-indexer" {
t.Fatalf("waits %v bodies %+v", f.waits(), f.bodies)
}
taken := func(branch string, source int64) {
raw, _ := json.Marshal(meshcatalog.TakeInTaken{Kind: "taken", Key: "m.1", Branch: branch, Source: source})
if err := f.p.w.taken.put(raw); err != nil {
t.Fatal(err)
}
}
taken("beta", 9) // another branch, and
taken(trunkBranch, 6)
f.pass()
if f.wantAt("asg") != orderFP("b") {
t.Fatal("a source revision before the trunk's head was taken in, and the assigner was wanted")
}
taken(trunkBranch, 7)
f.pass()
f.pass()
if f.wantAt("asg") != orderFP("a") || f.wantAt("gate") != orderFP("c") || len(f.waits()) != 0 {
t.Fatalf("after the head is taken in: asg %s gate %s waits %v", f.wantAt("asg"), f.wantAt("gate"), f.waits())
}
}
// A-260: an order wait still held after a restart is said again as wait-held, by a pass that remembers nothing, and ends at the next
// take-in.taken that names the trunk's head.
func TestAnOrderWaitHeldAfterARestartIsSaidAgain(t *testing.T) {
old, now := map[string]string{"asg": orderFP("a")}, map[string]string{"asg": orderFP("b")}
f := newOrderFixture(t, []orderMod{{name: "asg", run: orderFP("b"), trunk: orderFP("a"), holds: []string{mma.Seat}, src: 7}}, old, now)
f.pass()
f.pass()
if len(f.events) != 1 {
t.Fatalf("events %v", f.events)
}
f.p.orderSaid = nil // a process started again
f.pass()
if len(f.events) != 2 || f.events[1] != meshwants.EventWaitHeld || len(f.waits()) != 1 {
t.Fatalf("events %v waits %v", f.events, f.waits())
}
raw, _ := json.Marshal(meshcatalog.TakeInTaken{Kind: "taken", Key: "m.1", Branch: trunkBranch, Source: 8})
if err := f.p.w.taken.put(raw); err != nil {
t.Fatal(err)
}
f.pass()
f.pass()
if f.wantAt("asg") != orderFP("a") || len(f.waits()) != 0 || f.events[len(f.events)-1] != meshwants.EventWaitEnded {
t.Fatalf("asg at %s, waits %v, events %v", f.wantAt("asg"), f.waits(), f.events)
}
}
@@ -54,7 +54,7 @@ func (s *putTools) doWaits(args map[string]any) (any, error) {
if err != nil || !found {
continue
}
if acting.Is(key) {
if acting.Is(key) || acting.IsOrder(key) {
if a, err := acting.Decode(raw); err == nil && (arg(args, "node") == "" || a.Node == arg(args, "node")) && (arg(args, "module") == "" || a.Module == arg(args, "module")) {
said := actingSaid{Key: key, Wait: a}
if d, err := time.ParseDuration(a.Bound); err == nil && !a.Released {
@@ -146,9 +146,9 @@ func (s *putTools) doReleaseActing(node, module, why, by string) (any, error) {
w acting.Wait
}
var held []one
prefix := acting.Prefix + node + "." + module + "."
prefix, orderPrefix := acting.Prefix+node+"."+module+".", acting.OrderPrefix+node+"."+module+"."
for _, k := range keys {
if !strings.HasPrefix(k, prefix) {
if !strings.HasPrefix(k, prefix) && !strings.HasPrefix(k, orderPrefix) {
continue
}
raw, _, found, err := s.st.get(waitsState, k)
@@ -104,6 +104,11 @@ type passer struct {
publishWants publisher
// graph keeps the assignment graph revision the wants follow (graphkept.go, design 47 G §292, A-67); nil: none is kept.
graph *graphKeeper
// orders are §298's order edges computed in this pass, by their wait's key, and orderVisited the modules they were computed for (orderwaits.go).
orders map[string]orderEdge
orderVisited map[string]bool
// orderSaid are the order waits whose wait-held this process said.
orderSaid map[string]bool
// entries are the catalog entries by fingerprint asked of the seat mesh-catalog's verb `entry` (design 47 G §614, A-217): a
// held revert's running entry and its want's (revert.go); nil: no revert is asked about.
entries *meshcatalogclient.Entries
@@ -291,7 +296,9 @@ func (p *passer) wants(ctx context.Context) {
// Where grants are read from, once for the pass: the grants part's (issue 774, ADR 0571 clause 9).
grants := p.w.grantsBucket()
released := p.actingReleased()
p.orders, p.orderVisited = map[string]orderEdge{}, map[string]bool{}
defer func() { p.actingWaits(ctx, viewOf(trunkBranch).nodes, wanted) }()
defer func() { p.orderWaits(ctx, viewOf(trunkBranch).nodes, wanted) }()
for name := range names {
groups := p.groups(name, nodesOf[name], assignedBranches[name])
for gi, g := range groups {
@@ -358,7 +365,9 @@ func (p *passer) wants(ctx context.Context) {
m.MachineHeld = p.outside.Held
nodes = p.outsideReview(ctx, name, fp, m.Nodes, nodes, manifests[name].ClaimsSeat(mcseat.Seat), p.movedTo(wanted, nodes))
}
m.Acting = p.actingOf(name, released)
unread := p.orderEdgesOf(ctx, m, wantedOf[name], trunk)
actingOf, orderOf := p.actingOf(name, released), p.orderActing(name, released, unread, func() assign.Nodes { return nodes })
m.Acting = func(node string) []assign.ActingEdge { return append(actingOf(node), orderOf(node)...) }
out := assign.Wants(m, nodes, wantedOf[name], last)
for _, node := range m.Nodes {
for _, prov := range m.MissingProvisions(node) {
@@ -9,6 +9,7 @@ import (
"git.novox.be/novox/mesh-delivery/modules/mesh-delivery-assigner/asks"
mcc "git.novox.be/novox/mesh-delivery/modules/mesh-delivery-contracts/go/clients/meshcatalogclient"
mwc "git.novox.be/novox/mesh-delivery/modules/mesh-delivery-contracts/go/clients/meshwantsclient"
"git.novox.be/novox/mesh-sdk/go/catalogue"
"git.novox.be/novox/mesh-sdk/go/views"
"mesh-delivery-assigner/internal/assign"
@@ -108,16 +109,10 @@ func (p *passer) revertCrosses(ctx context.Context, node, module, wantFp string)
if !reverted {
return "", "", false
}
running, rs := p.entries.Need(ctx, run.Fingerprint)
applied, as := p.entries.Need(ctx, wantFp)
if rs != mcc.EntryAnswered || as != mcc.EntryAnswered {
changes, known := p.formChange(ctx, module, run.Fingerprint, wantFp)
if !known {
return "", "", false // asked: its answer wakes a pass
}
changes, err := assigngraph.FormChanges(running, applied)
if err != nil {
p.logf("the forms of %s's kept data between %s and %s are not read: %v", module, run.Fingerprint, wantFp, err)
return "", "", false
}
var names, forms []string
for _, c := range changes {
f := 0
@@ -158,3 +153,25 @@ func (p *passer) revertFinding(o assign.Observed, raised bool) {
p.said[said] = true
delete(p.said, map[bool]string{true: key + ".cleared", false: key + ".raised"}[raised])
}
// formChange reads the kept data whose form differs between module's entry at runFp and at wantFp (design 47 G §295 to §298, A-217), as
// the catalog seat's verb `entry` answers the two entries; known is false while either is asked, and woken by its answer.
func (p *passer) formChange(ctx context.Context, module, runFp, wantFp string) (changes []catalogue.FormChange, known bool) {
if p.entries == nil {
return nil, false
}
running, rs := p.entries.Need(ctx, runFp)
applied, as := p.entries.Need(ctx, wantFp)
if rs == mcc.EntryAsked || as == mcc.EntryAsked {
return nil, false
}
if rs != mcc.EntryAnswered || as != mcc.EntryAnswered {
return nil, true // the catalog holds no such entry: no kept data's form is read from it, so none changes
}
changes, err := assigngraph.FormChanges(running, applied)
if err != nil {
p.logf("the forms of %s's kept data between %s and %s are not read: %v", module, runFp, wantFp, err)
return nil, false
}
return changes, true
}
@@ -45,6 +45,8 @@ type world struct {
ends builtEnds
// released are the catalog's version-released heard, by module and source revision (buildended.go).
released releasedVersions
// taken is the newest source revision of each watched branch the indexer's take-in.taken named, as heard (orderwaits.go, §298, A-260).
taken takenSources
// 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
+1
View File
@@ -31,6 +31,7 @@ require git.novox.be/novox/mesh-delivery/modules/mesh-delivery-sdk v0.0.0-000101
require (
git.novox.be/novox/mesh-contracts/go/clients/nodeengineclient v0.0.0-00010101000000-000000000000
git.novox.be/novox/mesh-contracts/go/seats/meshbroker v0.0.0-00010101000000-000000000000
git.novox.be/novox/mesh-contracts/go/seats/nodeengine v0.0.0-00010101000000-000000000000
git.novox.be/novox/mesh-contracts/operator-and-conversation/go/clients/operatorchannelclient v0.0.0-00010101000000-000000000000
git.novox.be/novox/mesh-contracts/operator-and-conversation/go/seats/operatorchannel v0.0.0-00010101000000-000000000000
@@ -27,6 +27,21 @@ const (
// Prefix opens the key of an acting wait in the state, apart from the outside waits' `<node>.<module>.<fingerprint hex>`.
const Prefix = "acting."
// OrderPrefix opens the key of an order wait (design 47 G §298, A-259): a want held for the order of a delivery that crosses a form change,
// kept beside the acting waits in the state `waits`, as `order.<node>.<module>.<part>`.
const OrderPrefix = "order."
// The three order edges of §298, as a kept wait's Order names them.
const (
// OrderBackup: the new versions of the node-backup holder and of the holder of the bus module's seat run before a module whose kept
// data's form changes is moved.
OrderBackup = "backup-and-broker-first"
// OrderEngineLast: at a revert a node's node-engine goes back last on that node.
OrderEngineLast = "node-engine-last"
// OrderTakenIn: the reverted assigner and gate are wanted only once the reverted indexer took in the trunk's head.
OrderTakenIn = "taken-in-first"
)
// Held says whether the assigner holds the entries of step: composing and placing (§276's table).
func Held(step string) bool { return step == seatdef.StepCompose || step == seatdef.StepPlace }
@@ -44,6 +59,8 @@ type Wait struct {
ClockFrom time.Time `json:"clock-from"`
Bound string `json:"bound"`
// Released is a release by hand: the entry moves on with the part's old version; EndSaid says wait-ended was said for it.
// Order is which of §298's edges the wait stands on; empty for an acting wait.
Order string `json:"order,omitempty"`
Released bool `json:"released,omitempty"`
EndSaid bool `json:"end-said,omitempty"`
By string `json:"by,omitempty"`
@@ -53,6 +70,14 @@ type Wait struct {
// Key is the key of a wait in the state.
func Key(node, module, part string) string { return Prefix + node + "." + module + "." + part }
// OrderKey is the key of an order wait in the state.
func OrderKey(node, module, part string) string {
return OrderPrefix + node + "." + module + "." + part
}
// IsOrder says a kept value is an order wait's record.
func IsOrder(key string) bool { return strings.HasPrefix(key, OrderPrefix) }
// Is says a kept value is an acting wait's record.
func Is(key string) bool { return strings.HasPrefix(key, Prefix) }
@@ -92,7 +117,19 @@ func Pending(edges []meshwants.Edge, runs func(e meshwants.Edge) bool) []meshwan
func Body(w Wait, owner string) meshwants.Wait {
since, from := w.Since.UTC(), w.ClockFrom.UTC()
return meshwants.Wait{Step: w.Step, Owner: owner, Node: w.Node, Module: w.Module, Part: w.Part, Fp: w.Fp, Since: since, ClockFrom: &from, Bound: w.Bound,
Words: fmt.Sprintf("%s on %s waits in %s for %s at %s to run on %s", w.Module, w.Node, w.Step, w.Part, short(w.Fp), w.PartNode)}
Words: words(w)}
}
func words(w Wait) string {
switch w.Order {
case OrderBackup:
return fmt.Sprintf("%s on %s moves its kept data across a form change: it waits in %s for %s at %s to run on %s, which the restore point needs", w.Module, w.Node, w.Step, w.Part, short(w.Fp), w.PartNode)
case OrderEngineLast:
return fmt.Sprintf("%s on %s goes back last on its node: it waits in %s for %s at %s to run on %s, restored", w.Module, w.Node, w.Step, w.Part, short(w.Fp), w.PartNode)
case OrderTakenIn:
return fmt.Sprintf("%s on %s goes back only once the indexer took in the trunk's head: it waits in %s for source revision %s to be taken in", w.Module, w.Node, w.Step, w.Fp)
}
return fmt.Sprintf("%s on %s waits in %s for %s at %s to run on %s", w.Module, w.Node, w.Step, w.Part, short(w.Fp), w.PartNode)
}
func short(fp string) string {
@@ -130,7 +130,7 @@ func TestManifest(t *testing.T) {
}
// ADR 0618 §4: and it hears each node's engine say its reads changed; ADR 0636 §4 (issue 1966): and the seat list's two events;
// issue 1966 note 151: and the teller's ask-ended, its subject the one the assigner hears.
if !slices.Equal(m.Consumes, []string{"mesh-conditions.findings-asked", "mesh-conditions.condition-raised", "mesh-conditions.condition-changed", "mesh-conditions.condition-cleared", "mesh-catalog.catalog-revision", "mesh-catalog.registration-changed", "mesh-catalog.version-released", "mesh-catalog.seat-definition-changed", "mesh-catalog.seat-definition-removed", builds.BuilderModule + "." + builds.BuildEndedEvent, buckets.TellerModule + "." + tellerevents.AskEnded, "mesh-pipeline-runner.pipeline-ended", "mesh-pipeline-runner.group-ended", "node-engine.manifest-reads-changed", "node-engine.running", "mesh-delivery.pinned", "mesh-delivery.freed", "mesh-delivery.heads-replaced", "mesh-delivery.pins-stated"}) {
if !slices.Equal(m.Consumes, []string{"mesh-conditions.findings-asked", "mesh-conditions.condition-raised", "mesh-conditions.condition-changed", "mesh-conditions.condition-cleared", "mesh-catalog.catalog-revision", "mesh-catalog.registration-changed", "mesh-catalog.version-released", "mesh-catalog.seat-definition-changed", "mesh-catalog.seat-definition-removed", builds.BuilderModule + "." + builds.BuildEndedEvent, buckets.IndexerModule + ".take-in.taken", buckets.TellerModule + "." + tellerevents.AskEnded, "mesh-pipeline-runner.pipeline-ended", "mesh-pipeline-runner.group-ended", "node-engine.manifest-reads-changed", "node-engine.running", "mesh-delivery.pinned", "mesh-delivery.freed", "mesh-delivery.heads-replaced", "mesh-delivery.pins-stated"}) {
t.Errorf("consumes = %v, want the catalog half's revisions (design 47 G §195), the teller's ask-ended, the engines' reads changed and what they run", m.Consumes)
}
if want := []string{"mesh-delivery-indexer.manifests", "mesh-delivery-indexer.modules",
+2 -1
View File
@@ -1,6 +1,6 @@
{
"module": "mesh-delivery-assigner",
"version": "0.2",
"version": "0.3",
"slug": "assigner",
"runs-as": "mesh-delivery-assigner",
"claims": [
@@ -54,6 +54,7 @@
"mesh-catalog.seat-definition-changed",
"mesh-catalog.seat-definition-removed",
"mesh-delivery-builder.build.ended",
"mesh-delivery-indexer.take-in.taken",
"mesh-delivery-teller.ask-ended",
"mesh-pipeline-runner.pipeline-ended",
"mesh-pipeline-runner.group-ended",