diff --git a/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/actingwaits.go b/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/actingwaits.go new file mode 100644 index 00000000..8df05f4e --- /dev/null +++ b/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/actingwaits.go @@ -0,0 +1,152 @@ +package main + +import ( + "context" + + meshwants "git.novox.be/novox/mesh-delivery/modules/mesh-delivery-contracts/go/clients/meshwantsclient" + seat "git.novox.be/novox/mesh-delivery/modules/mesh-delivery-contracts/go/seats/meshwants" + + "mesh-delivery-assigner/internal/acting" + "mesh-delivery-assigner/internal/assign" + "mesh-delivery-assigner/internal/assignment" + "mesh-delivery-assigner/internal/buckets" +) + +// The waits for an acting part's new version in the steps the assigner holds (design 47 G §266, §268, §276): an entry of the assignment +// graph whose acting edge names composing or placing waits, its want held, until the part's version runs on the edge's node. The wait is +// kept in the state `waits` under `acting...`, said as `wait-held` when it starts and `wait-ended` (ran, released, +// replaced) when it ends; `assigner_waits` reads it and `assigner_release` with how `now` lets the entry move with the part's running +// version. A node away starts no wait: its clock starts when the node is present (§268). + +// actingOf is the Module.Acting of module: the held acting edges of its entry on a node, those whose wait was not released left out. +func (p *passer) actingOf(module string, released map[string]bool) func(node string) []assign.ActingEdge { + return func(node string) []assign.ActingEdge { + var out []assign.ActingEdge + for _, e := range p.entryEdges(node, module) { + if acting.Held(e.Step) && !released[acting.Key(node, module, e.Module)+"@"+e.Fp] { + out = append(out, assign.ActingEdge{Node: e.Node, Part: e.Module, Fp: e.Fp, Step: e.Step}) + } + } + return out + } +} + +// entryEdges are the acting edges of the entry of module on node in the revision the pass computed. +func (p *passer) entryEdges(node, module string) []seat.Edge { + g := p.graph + if g == nil { + return nil + } + for _, ref := range g.computed.Revision.Entries { + if ref.Node == node && ref.Module == module { + var out []seat.Edge + for _, e := range g.computed.Entries[ref.Digest].Edges { + if e.Kind == "acting" { + out = append(out, e) + } + } + return out + } + } + return nil +} + +// actingReleased is the waits released by hand and not yet run, by `@`. +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) { + out[key+"@"+w.Fp] = true + } + } + return out +} + +// actingWaits keeps the acting waits after the pass's wants: it starts the wait of each entry held for a part's version, ends the +// wait of one that ran, was replaced or was released, and says each. +func (p *passer) actingWaits(ctx context.Context, n assign.Nodes, wanted map[string]map[string]meshwants.Want) { + g := p.graph + if g == nil || p.kvs[buckets.Waits] == nil { + return + } + now := p.clock() + kept := p.w.read(buckets.Waits) + standing := map[string]bool{} + for _, ref := range g.computed.Revision.Entries { + entry := g.computed.Entries[ref.Digest] + if w, ok := wanted[ref.Module][ref.Node]; ok && w.Fp == entry.Fp { + continue // the entry's version is wanted already: nothing moves, so nothing waits + } + for _, e := range entry.Edges { + if e.Kind != "acting" || !acting.Held(e.Step) { + continue + } + key := acting.Key(ref.Node, ref.Module, e.Module) + if n.Runs(assign.ActingEdge{Node: e.Node, Part: e.Module, Fp: e.Fp, Step: e.Step}) { + continue + } + standing[key] = true + if raw := kept[key]; raw != nil { + if w, err := acting.Decode(raw); err == nil && w.Entry == entry.Fp && w.Fp == e.Fp { + p.actingReleaseSaid(ctx, key, w) + continue + } + } + if !n.Online[ref.Node] { + continue // a node away holds only itself, and its clock has not started (§268) + } + w := acting.Wait{Step: e.Step, Node: ref.Node, Module: ref.Module, Part: e.Module, PartNode: e.Node, Fp: e.Fp, Entry: entry.Fp, + Since: now, ClockFrom: now, Bound: acting.BoundOf(e).String()} + if err := p.wantsSay(seat.EventWaitHeld, acting.Body(w, assignment.Module), key+"."+e.Fp+"."+entry.Fp); err != nil { + p.logf("the wait of %s on %s for %s is not said, and is said at the next pass: %v", ref.Module, ref.Node, e.Module, err) + continue + } + p.put(ctx, buckets.Waits, key, func() ([]byte, error) { return acting.Encode(w) }) + } + } + for key, raw := range kept { + if !acting.Is(key) || standing[key] { + continue + } + w, err := acting.Decode(raw) + if err != nil { + continue + } + how := acting.HowRan + if w.Released { + how = acting.HowReleased + } else if e, ok := p.edgeOf(w); !ok || !n.Runs(e) { + 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) + } +} + +// actingReleaseSaid says wait-ended of a wait released by hand, once, and keeps its record so the entry is not held again. +func (p *passer) actingReleaseSaid(ctx context.Context, key string, w acting.Wait) { + if !w.Released || w.EndSaid { + return + } + body := acting.Body(w, assignment.Module) + body.How = acting.HowReleased + body.By = w.By + if err := p.wantsSay(seat.EventWaitEnded, body, key+"."+w.Fp+"."+w.Entry+"."+acting.HowReleased); err != nil { + p.logf("the release 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) + return + } + w.EndSaid = true + p.put(ctx, buckets.Waits, key, func() ([]byte, error) { return acting.Encode(w) }) +} + +// edgeOf is the edge a kept wait stands on, as an acting edge. +func (p *passer) edgeOf(w acting.Wait) (assign.ActingEdge, bool) { + return assign.ActingEdge{Node: w.PartNode, Part: w.Part, Fp: w.Fp, Step: w.Step}, true +} diff --git a/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/actingwaits_test.go b/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/actingwaits_test.go new file mode 100644 index 00000000..0f45a224 --- /dev/null +++ b/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/actingwaits_test.go @@ -0,0 +1,118 @@ +package main + +import ( + "context" + "encoding/json" + "strings" + "testing" + "time" + + "git.novox.be/novox/mesh-contracts/go/seats/nodeengine" + mwc "git.novox.be/novox/mesh-delivery/modules/mesh-delivery-contracts/go/clients/meshwantsclient" + meshwants "git.novox.be/novox/mesh-delivery/modules/mesh-delivery-contracts/go/seats/meshwants" + + "mesh-delivery-assigner/internal/acting" + "mesh-delivery-assigner/internal/assign" + "mesh-delivery-assigner/internal/buckets" +) + +// Design 47 G §266, §268, §276 (AS9): an entry with an acting edge in composing waits for the part's version to run on the edge's node; +// the wait is kept, said as wait-held with its bound and clock-from, read by assigner_waits, released by assigner_release with how +// `now` and ended when the version runs. +func TestAnEntryWaitsForItsActingPartAndTheWaitIsKeptSaidAndEnded(t *testing.T) { + now := time.Date(2026, 11, 12, 9, 0, 0, 0, time.UTC) + p := newPasser(kvSet{buckets.Waits: &memKV{values: map[string][]byte{}}}, now) + var events []string + var bodies []meshwants.Wait + p.publishWants = func(event string, body any, id string) error { + events = append(events, event) + bodies = append(bodies, body.(meshwants.Wait)) + return nil + } + entry := meshwants.AssignmentEntry{Node: "ace", Module: "nats", Fp: fpTest, Edges: []meshwants.Edge{ + {Kind: "acting", Node: "ace", Module: "mesh-host", Fp: "src2:new", Seat: "node-engine", Step: "compose", Bound: "15m"}, + {Kind: "acting", Node: "ace", Module: "indexer", Fp: "src2:idx", Seat: "mesh-catalog", Step: "take-in"}}} + p.graph = &graphKeeper{} + p.graph.computed.Revision.Entries = []meshwants.NodeEntryRef{{Node: "ace", Module: "nats", Digest: "d1"}} + p.graph.computed.Entries = map[string]meshwants.AssignmentEntry{"d1": entry} + n := assign.Nodes{Online: map[string]bool{"ace": true}, Running: map[string]map[string]nodeengine.Running{"ace": {"mesh-host": {Fingerprint: "src2:old"}}}} + wanted := map[string]map[string]mwc.Want{"nats": {"ace": {Fp: "src2:older"}}} + ctx := context.Background() + + if got := p.actingOf("nats", nil)("ace"); len(got) != 1 || got[0].Part != "mesh-host" { + t.Fatalf("the assigner holds composing only: %+v", got) + } + p.actingWaits(ctx, n, wanted) + p.actingWaits(ctx, n, wanted) // said once + if len(events) != 1 || events[0] != meshwants.EventWaitHeld { + t.Fatalf("events %v", events) + } + b := bodies[0] + if b.Step != "compose" || b.Part != "mesh-host" || b.Fp != "src2:new" || b.Bound != "15m0s" || b.ClockFrom == nil || !b.ClockFrom.Equal(now) || b.Owner == "" { + t.Errorf("wait-held %+v", b) + } + key := acting.Key("ace", "nats", "mesh-host") + if _, err := acting.Decode(p.w.read(buckets.Waits)[key]); err != nil { + t.Fatalf("the wait is kept: %v", err) + } + // a release by hand lets the entry move, and ends the wait once + w, _ := acting.Decode(p.w.read(buckets.Waits)[key]) + w.Released = true + raw, _ := acting.Encode(w) + p.w.put(buckets.Waits, key, raw) + if got := p.actingOf("nats", p.actingReleased())("ace"); len(got) != 0 { + t.Errorf("a released wait still holds: %+v", got) + } + p.actingWaits(ctx, n, wanted) + p.actingWaits(ctx, n, wanted) + if len(events) != 2 || events[1] != meshwants.EventWaitEnded || bodies[1].How != acting.HowReleased { + t.Fatalf("events %v bodies %+v", events, bodies) + } + // the version runs: the record goes, said nothing more + n.Running["ace"]["mesh-host"] = nodeengine.Running{Fingerprint: "src2:new"} + p.actingWaits(ctx, n, wanted) + if len(p.w.read(buckets.Waits)) != 0 || len(events) != 2 { + t.Errorf("kept %v events %v", p.w.read(buckets.Waits), events) + } +} + +func TestAnActingWaitOfANodeAwayStartsNoClockAndAnEntryWantedAlreadyWaitsNot(t *testing.T) { + p := newPasser(kvSet{buckets.Waits: &memKV{values: map[string][]byte{}}}, time.Now()) + p.publishWants = func(string, any, string) error { t.Fatal("said"); return nil } + p.graph = &graphKeeper{} + p.graph.computed.Revision.Entries = []meshwants.NodeEntryRef{{Node: "ace", Module: "nats", Digest: "d1"}} + p.graph.computed.Entries = map[string]meshwants.AssignmentEntry{"d1": {Node: "ace", Module: "nats", Fp: "src2:a", Edges: []meshwants.Edge{ + {Kind: "acting", Node: "ace", Module: "mesh-host", Fp: "src2:new", Step: "compose"}}}} + n := assign.Nodes{Online: map[string]bool{}} + p.actingWaits(context.Background(), n, map[string]map[string]mwc.Want{}) + n.Online["ace"] = true + p.actingWaits(context.Background(), n, map[string]map[string]mwc.Want{"nats": {"ace": {Fp: "src2:a"}}}) + if len(p.w.read(buckets.Waits)) != 0 || strings.Contains(p.graph.computed.Entries["d1"].Fp, "x") { + t.Errorf("kept %v", p.w.read(buckets.Waits)) + } +} + +func TestTheToolsReadAndReleaseAnActingWait(t *testing.T) { + var acts []string + s := newPutTools(memStore{waitsState: &memKV{values: map[string][]byte{}}}, func(key string, body any) (json.RawMessage, error) { + acts = append(acts, key) + return json.RawMessage(`{"ok":true,"log":"logged"}`), nil + }) + w := acting.Wait{Step: "compose", Node: "ace", Module: "nats", Part: "mesh-host", PartNode: "ace", Fp: "src2:new", Entry: "src2:a", Since: time.Now(), ClockFrom: time.Now(), Bound: "30m0s"} + raw, _ := acting.Encode(w) + s.st.put(waitsState, acting.Key("ace", "nats", "mesh-host"), raw) + got, err := s.doWaits(map[string]any{"module": "nats"}) + if list := got.(map[string]any)["waits"].([]any); err != nil || len(list) != 1 || list[0].(actingSaid).BoundEnds == "" { + t.Fatalf("assigner_waits = %+v %v", got, err) + } + if _, err := s.doRelease(map[string]any{"node": "ace", "module": "nats", "how": "now"}, "x"); err == nil { + t.Error("a release with no why is taken") + } + if _, err := s.doRelease(map[string]any{"node": "ace", "module": "nats", "how": "now", "why": "the new engine is known good"}, "shanks.node-tools"); err != nil || len(acts) == 0 { + t.Fatalf("%v %v", err, acts) + } + raw, _, _, _ = s.st.get(waitsState, acting.Key("ace", "nats", "mesh-host")) + if r, _ := acting.Decode(raw); !r.Released || !strings.Contains(r.By, "known good") { + t.Errorf("wait = %+v", r) + } +} diff --git a/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/graphkept.go b/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/graphkept.go index 884723dd..bfa522b0 100644 --- a/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/graphkept.go +++ b/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/graphkept.go @@ -36,6 +36,8 @@ type graphKeeper struct { said string // pins removes a revision that ceased to be current once nothing pins it (pins.go); nil keeps every revision. pins *pinKeeper + // computed is the revision computed by the last pass, with its entries: the edges the wants wait on (actingwaits.go). + computed assigngraph.Computed } // keepGraph computes the revision of this pass and keeps it when the pass's inputs differ from the last computed. @@ -51,6 +53,7 @@ func (p *passer) keepGraph(ctx context.Context) { in.FirstBranch = func(module string) string { return p.ends.branchOf(module, "") } } c := assigngraph.Compute(in) + g.computed = c for _, u := range c.Uncomputed { if g.said != u { g.said = u diff --git a/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/outside_test.go b/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/outside_test.go index 4a06236e..9623e4f1 100644 --- a/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/outside_test.go +++ b/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/outside_test.go @@ -265,10 +265,10 @@ func TestTheToolsReadAndReleaseAnOutsideWait(t *testing.T) { s.st.put(waitsState, k, v) } got, err := s.doWaits(map[string]any{"module": "nats"}) - if err != nil || len(got.(map[string]any)["waits"].([]waitSaid)) != 1 { + if err != nil || len(got.(map[string]any)["waits"].([]any)) != 1 { t.Fatalf("assigner_waits = %+v %v", got, err) } - if w := got.(map[string]any)["waits"].([]waitSaid)[0]; w.On != "unreachable:hub" || w.BoundEnds == "" { + if w := got.(map[string]any)["waits"].([]any)[0].(waitSaid); w.On != "unreachable:hub" || w.BoundEnds == "" { t.Errorf("wait = %+v", w) } if _, err := s.doRelease(map[string]any{"node": "ace", "module": "nats", "how": "sideways", "why": "x"}, "shanks.node-tools"); err == nil { 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 9d22b933..ae75ce2f 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 @@ -11,6 +11,7 @@ import ( "git.novox.be/novox/mesh-contracts/health-and-repair/go/clients/meshhandactsclient" stdio "git.novox.be/novox/mesh-sdk/go" + "mesh-delivery-assigner/internal/acting" "mesh-delivery-assigner/internal/buckets" "mesh-delivery-assigner/internal/outside" ) @@ -33,6 +34,13 @@ type waitSaid struct { BoundEnds string `json:"bound-ends,omitempty"` } +// actingSaid is one wait for an acting part's new version as assigner_waits answers it (actingwaits.go). +type actingSaid struct { + Key string `json:"key"` + acting.Wait + BoundEnds string `json:"bound-ends,omitempty"` +} + // doWaits is the tool assigner_waits. func (s *putTools) doWaits(args map[string]any) (any, error) { keys, err := s.st.keys(waitsState) @@ -40,12 +48,22 @@ func (s *putTools) doWaits(args map[string]any) (any, error) { return nil, fmt.Errorf("the waits cannot be listed: %w", err) } sort.Strings(keys) - out := []waitSaid{} + out := []any{} for _, key := range keys { raw, _, found, err := s.st.get(waitsState, key) if err != nil || !found { continue } + if acting.Is(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 { + said.BoundEnds = a.ClockFrom.Add(d).UTC().Format("2006-01-02T15:04:05Z") + } + out = append(out, said) + } + continue + } w, err := outside.Decode(raw) if err != nil || (arg(args, "node") != "" && w.Node != arg(args, "node")) || (arg(args, "module") != "" && w.Module != arg(args, "module")) { continue @@ -65,6 +83,9 @@ func (s *putTools) doWaits(args map[string]any) (any, error) { // module, which the assigner's process takes at its next pass. func (s *putTools) doReleaseWait(args map[string]any, by string) (any, error) { node, module, how, why := arg(args, "node"), arg(args, "module"), arg(args, "how"), arg(args, "why") + if how == acting.HowNow { + return s.doReleaseActing(node, module, why, by) + } if !slices.Contains([]string{outside.HowGoOn, outside.HowFailed, outside.HowRejudge}, how) { return nil, fmt.Errorf("how is %s, %s or %s, not %q", outside.HowGoOn, outside.HowFailed, outside.HowRejudge, how) } @@ -110,9 +131,61 @@ func (s *putTools) doReleaseWait(args map[string]any, by string) (any, error) { return map[string]any{"released": key, "how": how, "by": by, "hand-act": taken.Log}, nil } +// doReleaseActing is assigner_release with how `now`: the entry of module on node moves with its acting parts' running versions. The +// hand act is recorded first; the release is written on each standing wait of the entry, and the process says wait-ended at its next pass. +func (s *putTools) doReleaseActing(node, module, why, by string) (any, error) { + if why == "" { + return nil, errors.New("release says why: it is kept in the hand-act log") + } + keys, err := s.st.keys(waitsState) + if err != nil { + return nil, fmt.Errorf("the waits cannot be listed: %w", err) + } + type one struct { + key string + w acting.Wait + } + var held []one + prefix := acting.Prefix + node + "." + module + "." + for _, k := range keys { + if !strings.HasPrefix(k, prefix) { + continue + } + raw, _, found, err := s.st.get(waitsState, k) + if err != nil || !found { + continue + } + if w, err := acting.Decode(raw); err == nil && !w.Released { + held = append(held, one{k, w}) + } + } + if len(held) == 0 { + return nil, fmt.Errorf("%s holds no wait of %s for an acting part: assigner_waits lists them", node, module) + } + rec := meshhandactsclient.Recorder{Ask: s.ask} + taken, err := rec.Record(meshhandactsclient.Request{What: fmt.Sprintf("assigner_release %s %s %s", node, module, acting.HowNow), Why: why, + Cause: "acting-wait", Verb: "mesh-wants.assigner_release"}) + if err != nil { + return nil, fmt.Errorf("the hand act could not be recorded, so nothing was released: %w", err) + } + released := []string{} + for _, h := range held { + h.w.Released, h.w.By, h.w.Why = true, by+": "+why, why + data, err := acting.Encode(h.w) + if err != nil { + return nil, err + } + if err := s.st.put(waitsState, h.key, data); err != nil { + return nil, fmt.Errorf("the release of %s on %s is not written: %w", module, node, err) + } + released = append(released, h.key) + } + return map[string]any{"released": released, "how": acting.HowNow, "by": by, "hand-act": taken.Log}, nil +} + // waitsTool is assigner_waits as the runtime serves it. func (s *putTools) waitsTool() stdio.Tool { - return stdio.Tool{Name: toolWaits, Description: "Every wait of the delivery on an outside cause and where it stands: a moved module's failing health check on a " + + return stdio.Tool{Name: toolWaits, Description: "Every wait of the delivery and where it stands: an entry waiting for an acting part's new version in composing or placing (§266, §268), with its part, version, step, clock-from and bound; and a moved module's failing health check on a " + "machine whose open conditions name a cause outside the delivery (design 47 G §706), with its module, machine, fingerprint, the " + "condition's key, what it stands `on`, its clock-from, bound, its ask and its state; a settle shows as settling with its end. Narrowed by " + "machine or module. Read-only.", 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 71581d0e..e0d97613 100644 --- a/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/pass.go +++ b/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/pass.go @@ -284,6 +284,8 @@ func (p *passer) wants(ctx context.Context) { defer func() { p.sayNeither(neither, stale, left) }() // 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() + defer func() { p.actingWaits(ctx, viewOf(trunkBranch).nodes, wanted) }() for name := range names { groups := p.groups(name, nodesOf[name], assignedBranches[name]) for gi, g := range groups { @@ -350,6 +352,7 @@ 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) 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/tools.go b/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/tools.go index a362a264..b8325a1d 100644 --- a/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/tools.go +++ b/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/tools.go @@ -366,7 +366,7 @@ func (s *putTools) tools() []stdio.Tool { "rules write the machine's want again from their next pass. Nothing changes on the machine until they move it.", Input: schema(map[string]string{"node": "the machine", "module": "the module", "why": "why, kept with the put", "how": "with how, the wait of the delivery on an outside cause (assigner_waits) is released, not a put: go-on (the module counts healthy there), " + - "failed (the failing check is the delivery's) or rejudge (judge it again now); written in the hand-act log first"}, + "failed (the failing check is the delivery's), rejudge (judge it again now), or now (the entry waiting for an acting part's new version moves with the running one); written in the hand-act log first"}, []string{"node", "module", "why"}), Run: func(a map[string]any) (any, error) { return s.doRelease(a, stdio.CallerOf(a)) }}, s.waitsTool(), diff --git a/modules/mesh-delivery-assigner/internal/acting/acting.go b/modules/mesh-delivery-assigner/internal/acting/acting.go new file mode 100644 index 00000000..e4607f99 --- /dev/null +++ b/modules/mesh-delivery-assigner/internal/acting/acting.go @@ -0,0 +1,103 @@ +// Package acting is the assigner's wait for an acting part's new version (design 47 G §266, §268, §276): an entry of the assignment +// graph whose edge of kind acting names a step the assigner holds (composing, placing) waits until the acting part's version runs +// on the node of the edge, with the bound of the rule that gave the edge, or the step's default. The package holds the rules and +// the kept record; the assigner's process reads what runs, says the events and keeps the records in its state `waits`. +package acting + +import ( + "encoding/json" + "fmt" + "strings" + "time" + + "git.novox.be/novox/mesh-delivery/modules/mesh-delivery-contracts/go/seats/meshwants" + "git.novox.be/novox/mesh-sdk/go/seatdef" +) + +// HowNow is the `how` of assigner_release for a wait for an acting part: the entry is wanted now, with the version running. +const HowNow = "now" + +// The ways a wait ends, as `wait-ended` says them. +const ( + HowRan = "ran" + HowReleased = "released" + HowReplaced = "replaced" +) + +// Prefix opens the key of an acting wait in the state, apart from the outside waits' `..`. +const Prefix = "acting." + +// 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 } + +// Wait is one wait as kept: Module's entry on Node waits for Part's version Fp to run on PartNode, in Step. +type Wait struct { + V int `json:"v"` + Step string `json:"step"` + Node string `json:"node"` + Module string `json:"module"` + Part string `json:"part"` + PartNode string `json:"part-node"` + Fp string `json:"fp"` + Entry string `json:"entry"` + Since time.Time `json:"since"` + 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. + Released bool `json:"released,omitempty"` + EndSaid bool `json:"end-said,omitempty"` + By string `json:"by,omitempty"` + Why string `json:"why,omitempty"` +} + +// Key is the key of a wait in the state. +func Key(node, module, part string) string { return Prefix + node + "." + module + "." + part } + +// Is says a kept value is an acting wait's record. +func Is(key string) bool { return strings.HasPrefix(key, Prefix) } + +// Encode is w as JSON. +func Encode(w Wait) ([]byte, error) { w.V = 1; return json.Marshal(w) } + +// Decode reads a kept wait. +func Decode(raw []byte) (Wait, error) { + var w Wait + if err := json.Unmarshal(raw, &w); err != nil { + return w, err + } + if w.V != 1 || w.Step == "" || w.Node == "" || w.Module == "" || w.Part == "" { + return w, fmt.Errorf("acting: %q is not a kept acting wait", raw) + } + return w, nil +} + +// BoundOf is the bound of the edge's wait: its rule's, else its step's default (seatdef.ActingRule.Wait). +func BoundOf(e meshwants.Edge) time.Duration { + return seatdef.ActingRule{Step: e.Step, Bound: e.Bound}.Wait() +} + +// Pending are the acting edges of the assigner's steps whose part's version does not run yet: runs says the edge's part runs +// its version on the edge's node. +func Pending(edges []meshwants.Edge, runs func(e meshwants.Edge) bool) []meshwants.Edge { + var out []meshwants.Edge + for _, e := range edges { + if e.Kind == "acting" && Held(e.Step) && !runs(e) { + out = append(out, e) + } + } + return out +} + +// Body is the wait as the seat mesh-wants says it. +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)} +} + +func short(fp string) string { + if len(fp) > 19 { + return fp[:19] + } + return fp +} diff --git a/modules/mesh-delivery-assigner/internal/assign/wants.go b/modules/mesh-delivery-assigner/internal/assign/wants.go index d3cb0742..e1decc0b 100644 --- a/modules/mesh-delivery-assigner/internal/assign/wants.go +++ b/modules/mesh-delivery-assigner/internal/assign/wants.go @@ -101,6 +101,36 @@ type Module struct { // MachineHeld says whether the node's own network condition is open, which holds the machine as a whole: no want moves to it // until that clears (design 47 G §706 *M's own network*). Nil: none is held. MachineHeld func(node string) bool + // Acting are the acting edges of the assignment graph entry of this module on node that the assigner holds (composing, placing) + // and whose wait was not released by hand (design 47 G §266, §268): the want moves only once each acting part's version runs + // on the edge's node, as Nodes.Running says it. Nil: none. + Acting func(node string) []ActingEdge +} + +// ActingEdge is one acting edge: Part's version Fp acts in Step on Node, before the entry moves. +type ActingEdge struct{ Node, Part, Fp, Step string } + +// Runs says the part's version of the edge runs on its node, as the node-engine seat's `running` says it. +func (n Nodes) Runs(e ActingEdge) bool { + if n.Running != nil { + r, ok := n.Running[e.Node][e.Part] + return ok && !r.Stopped && r.Fingerprint == e.Fp + } + h, ok := n.Haves[e.Node][e.Part] + return ok && healthy(h, e.Fp) +} + +// actingWaiting says why the module's want on node waits for an acting part's new version (§266): the first edge not run. +func (m Module) actingWaiting(node string, n Nodes) (string, bool) { + if m.Acting == nil { + return "", false + } + for _, e := range m.Acting(node) { + if !n.Runs(e) { + return fmt.Sprintf("%s on %s waits in %s for %s at %s to run on %s", m.Name, node, e.Step, e.Part, short(e.Fp), e.Node), true + } + } + return "", false } // WantParts is the capability a node-engine of G states (ADR 0614 §4, design 47 G §525): it reads a want's `parts` and `kept-on`. @@ -219,6 +249,9 @@ func holds(m Module, node string, n Nodes, ordering bool) (bool, int, string) { if _, ok := m.composition(node, m.FP); !ok { return false, mwc.RuleComposed, fmt.Sprintf("%s has named no composition of its own for %s at %s yet", node, m.Name, short(m.FP)) } + if words, waiting := m.actingWaiting(node, n); waiting { + return false, mwc.RuleComposed, words + } if words, waiting := m.provisionWaiting(node); waiting { return false, mwc.RuleProvisions, words } diff --git a/modules/mesh-delivery-assigner/internal/assign/wants_test.go b/modules/mesh-delivery-assigner/internal/assign/wants_test.go index d0f3be9f..e048eb0f 100644 --- a/modules/mesh-delivery-assigner/internal/assign/wants_test.go +++ b/modules/mesh-delivery-assigner/internal/assign/wants_test.go @@ -1608,3 +1608,28 @@ func TestAWantNamesAKeptBuildsPartsAndNodeForANodeThatReadsThem(t *testing.T) { } } } + +// Design 47 G §266, §268 (AS9): an entry whose acting edge names a step the assigner holds waits, its want held, until the acting part's +// version runs on the edge's node, as the node-engine seat's `running` says it; then the want moves. +func TestAWantWaitsForItsActingPartsNewVersion(t *testing.T) { + edge := ActingEdge{Node: "g14", Part: "mesh-host", Fp: fNew, Step: "compose"} + run := func(fp string, stopped bool) Nodes { + _, n := ready() + n.Running = map[string]map[string]nodeengine.Running{"g14": {"mesh-host": {Fingerprint: fp, Stopped: stopped}}} + return n + } + m, _ := ready() + m.Acting = func(string) []ActingEdge { return []ActingEdge{edge} } + m.Nodes = []string{"g14"} + out := Wants(m, run(fOld, false), wantedOld(), nil) + if out.Wants["g14"].Fp == fNew || out.Rules["g14"].Rule != mwc.RuleComposed || !strings.Contains(out.Rules["g14"].Words, "waits in compose for mesh-host") { + t.Fatalf("the old version runs: want %+v rule %+v", out.Wants["g14"], out.Rules["g14"]) + } + if out := Wants(m, run(fNew, false), wantedOld(), nil); out.Wants["g14"].Fp != fNew { + t.Errorf("the new version runs: want %+v", out.Wants["g14"]) + } + m.Acting = func(string) []ActingEdge { return nil } // a released wait is no edge + if out := Wants(m, run(fOld, false), wantedOld(), nil); out.Wants["g14"].Fp != fNew { + t.Errorf("released: want %+v", out.Wants["g14"]) + } +} diff --git a/modules/mesh-delivery-assigner/internal/outside/outside.go b/modules/mesh-delivery-assigner/internal/outside/outside.go index f7b4209e..64d81ebc 100644 --- a/modules/mesh-delivery-assigner/internal/outside/outside.go +++ b/modules/mesh-delivery-assigner/internal/outside/outside.go @@ -106,7 +106,9 @@ func OwnNetwork(conds []Cond) map[string]bool { // what ended it for the rest of that fingerprint on that node: `go-on` (counted healthy by the operator's word) or `failed` // (the delivery's failure), so a check that goes on failing is not waited on again. type Wait struct { - V int `json:"v"` + V int `json:"v"` + // Step is set only on a record of an acting wait (internal/acting), which this package never reads as its own. + Step string `json:"step,omitempty"` Node string `json:"node"` Module string `json:"module"` FP string `json:"fp"` @@ -141,7 +143,7 @@ func Decode(raw []byte) (Wait, error) { if err := json.Unmarshal(raw, &w); err != nil { return w, err } - if w.V != 1 || w.Node == "" || w.Module == "" { + if w.V != 1 || w.Step != "" || w.Node == "" || w.Module == "" { return w, fmt.Errorf("outside: %q is not a kept outside wait", raw) } return w, nil