From cd51fec81a97d538f609fb8f44e8fadf861a6206 Mon Sep 17 00:00:00 2001 From: jochen Date: Thu, 12 Nov 2026 09:49:25 +0100 Subject: [PATCH] =?UTF-8?q?B2l=20AS17=20(4):=20=C2=A7298's=20three=20order?= =?UTF-8?q?=20edges=20are=20acting=20waits=20of=20the=20assigner=20under?= =?UTF-8?q?=20`order.`?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 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 --- .../cmd/mesh-delivery-assigner/actingwaits.go | 2 +- .../cmd/mesh-delivery-assigner/buildended.go | 7 +- .../cmd/mesh-delivery-assigner/main.go | 2 +- .../cmd/mesh-delivery-assigner/orderwaits.go | 296 +++++++++++++++ .../mesh-delivery-assigner/orderwaits_test.go | 355 ++++++++++++++++++ .../mesh-delivery-assigner/outside_tools.go | 6 +- .../cmd/mesh-delivery-assigner/pass.go | 11 +- .../cmd/mesh-delivery-assigner/revert.go | 33 +- .../cmd/mesh-delivery-assigner/world.go | 2 + modules/mesh-delivery-assigner/go.mod | 1 + .../internal/acting/acting.go | 39 +- .../mesh-delivery-assigner/manifest_test.go | 2 +- modules/mesh-delivery-assigner/module.json | 3 +- 13 files changed, 740 insertions(+), 19 deletions(-) create mode 100644 modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/orderwaits.go create mode 100644 modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/orderwaits_test.go diff --git a/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/actingwaits.go b/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/actingwaits.go index 8df05f4e..4b7f7095 100644 --- a/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/actingwaits.go +++ b/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/actingwaits.go @@ -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 } } diff --git a/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/buildended.go b/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/buildended.go index 98c0465d..899c4ade 100644 --- a/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/buildended.go +++ b/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/buildended.go @@ -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)) } diff --git a/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/main.go b/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/main.go index d671bb71..c39440fe 100644 --- a/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/main.go +++ b/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/main.go @@ -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) diff --git a/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/orderwaits.go b/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/orderwaits.go new file mode 100644 index 00000000..2a921b0f --- /dev/null +++ b/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/orderwaits.go @@ -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...` 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) + } +} diff --git a/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/orderwaits_test.go b/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/orderwaits_test.go new file mode 100644 index 00000000..36b3835f --- /dev/null +++ b/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/orderwaits_test.go @@ -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) + } +} diff --git a/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/outside_tools.go b/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/outside_tools.go index ae75ce2f..461a9e01 100644 --- a/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/outside_tools.go +++ b/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/outside_tools.go @@ -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) diff --git a/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/pass.go b/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/pass.go index 9eb26f44..86b1e644 100644 --- a/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/pass.go +++ b/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/pass.go @@ -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) { diff --git a/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/revert.go b/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/revert.go index 80640eb7..5b61ea2d 100644 --- a/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/revert.go +++ b/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/revert.go @@ -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 +} diff --git a/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/world.go b/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/world.go index e591751b..b9a8a0d8 100644 --- a/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/world.go +++ b/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/world.go @@ -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 diff --git a/modules/mesh-delivery-assigner/go.mod b/modules/mesh-delivery-assigner/go.mod index 39fb4f5d..5addb9ec 100644 --- a/modules/mesh-delivery-assigner/go.mod +++ b/modules/mesh-delivery-assigner/go.mod @@ -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 diff --git a/modules/mesh-delivery-assigner/internal/acting/acting.go b/modules/mesh-delivery-assigner/internal/acting/acting.go index e4607f99..abc217c1 100644 --- a/modules/mesh-delivery-assigner/internal/acting/acting.go +++ b/modules/mesh-delivery-assigner/internal/acting/acting.go @@ -27,6 +27,21 @@ const ( // Prefix opens the key of an acting wait in the state, apart from the outside waits' `..`. 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...`. +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 { diff --git a/modules/mesh-delivery-assigner/manifest_test.go b/modules/mesh-delivery-assigner/manifest_test.go index f453d867..cc62a74b 100644 --- a/modules/mesh-delivery-assigner/manifest_test.go +++ b/modules/mesh-delivery-assigner/manifest_test.go @@ -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", diff --git a/modules/mesh-delivery-assigner/module.json b/modules/mesh-delivery-assigner/module.json index cc25c1c1..e2a37e59 100644 --- a/modules/mesh-delivery-assigner/module.json +++ b/modules/mesh-delivery-assigner/module.json @@ -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",