AS9: an entry waits for its acting part's new version in composing and placing; assigner_waits lists the wait, assigner_release now lets it go
Builds: design 47 G §266, §268, §276, §269 (the assigner's part: wait-held with bound and clock-from, wait-ended; acting-part-late is the watchman's) Claude-Session: https://claude.ai/code/session_01NcbgfGKaxw984XyAPfLm4h
This commit is contained in:
@@ -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.<node>.<module>.<part>`, 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 `<key>@<part's fingerprint>`.
|
||||
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
|
||||
}
|
||||
@@ -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)
|
||||
}
|
||||
}
|
||||
@@ -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
|
||||
|
||||
@@ -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 {
|
||||
|
||||
@@ -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.",
|
||||
|
||||
@@ -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) {
|
||||
|
||||
@@ -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(),
|
||||
|
||||
@@ -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' `<node>.<module>.<fingerprint hex>`.
|
||||
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
|
||||
}
|
||||
@@ -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
|
||||
}
|
||||
|
||||
@@ -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"])
|
||||
}
|
||||
}
|
||||
|
||||
@@ -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
|
||||
|
||||
Reference in New Issue
Block a user