Assigner: a moved module's failing check on an outside cause is a wait that ends by itself
The wait (step outside, on, bound, clock-from, the settle, conditions-unread), its events, finding and ask, assigner_waits and assigner_release with how, the settings, and a machine's own open network holding its wants. Builds: design 47 G §706 (assigner), §498, §276, §633, §700 Claude-Session: https://claude.ai/code/session_01NcbgfGKaxw984XyAPfLm4h
This commit is contained in:
@@ -38,6 +38,9 @@ const (
|
||||
// `assignments.<proposal id>`. The kind needs no proof, as the bus's put does not: its yes is an Approve, which only a
|
||||
// channel proving who answered takes and the view never does (ViewAnswerFits).
|
||||
AskKindAssignments = "assignments"
|
||||
// AskKindOutside is the kind of the assigner's ask about a moved module whose failing health check has waited past its bound
|
||||
// on an outside cause (design 47 G §706): go on, or count it as failed. `outside.<node>.<module>.<fingerprint hex>`.
|
||||
AskKindOutside = "outside"
|
||||
)
|
||||
|
||||
// AskKey is an ask's key read whole: its Kind, and the parts its kind has: a merge's Group and heads Hash, a hold's Module
|
||||
@@ -60,11 +63,11 @@ func AskNeedsProof(kind string) (bool, error) {
|
||||
switch kind {
|
||||
case meshdeliveryclient.AskKindMerge:
|
||||
return true, nil
|
||||
case AskKindHold, AskKindRestore, AskKindBackup, AskKindRecord, AskKindAssignments:
|
||||
case AskKindHold, AskKindRestore, AskKindBackup, AskKindRecord, AskKindAssignments, AskKindOutside:
|
||||
return false, nil
|
||||
}
|
||||
return false, fmt.Errorf("asks: %q is not a kind of ask, %s, %s, %s, %s, %s or %s", kind, meshdeliveryclient.AskKindMerge, AskKindHold,
|
||||
AskKindRestore, AskKindBackup, AskKindRecord, AskKindAssignments)
|
||||
return false, fmt.Errorf("asks: %q is not a kind of ask, %s, %s, %s, %s, %s, %s or %s", kind, meshdeliveryclient.AskKindMerge, AskKindHold,
|
||||
AskKindRestore, AskKindBackup, AskKindRecord, AskKindAssignments, AskKindOutside)
|
||||
}
|
||||
|
||||
// HoldAskKey is the key of the ask about a module's held fingerprint: `hold.<module>.<hex>`.
|
||||
@@ -89,6 +92,19 @@ func RestoreAskKey(node, module, fingerprint string) (string, error) {
|
||||
return AskKindRestore + "." + node + "." + key, nil
|
||||
}
|
||||
|
||||
// OutsideAskKey is the key of the ask about a module's failing check on a node that waited past its bound on an outside cause:
|
||||
// `outside.<node>.<module>.<hex>`.
|
||||
func OutsideAskKey(node, module, fingerprint string) (string, error) {
|
||||
key, err := RestoreAskKey(node, module, fingerprint)
|
||||
if err != nil {
|
||||
return "", err
|
||||
}
|
||||
return AskKindOutside + strings.TrimPrefix(key, AskKindRestore), nil
|
||||
}
|
||||
|
||||
// OutsideAbout is what the outside ask is about: `<node>.<module>.<hex>` (design 47 G §706).
|
||||
func OutsideAbout(key string) string { return strings.TrimPrefix(key, AskKindOutside+".") }
|
||||
|
||||
// BackupAskKey is the key of the ask about a module's failed backup on a node before it applies a fingerprint:
|
||||
// `backup.<node>.<module>.<hex>`.
|
||||
func BackupAskKey(node, module, fingerprint string) (string, error) {
|
||||
@@ -157,6 +173,13 @@ func ParseAnyAskKey(key string) (AskKey, error) {
|
||||
return refused(err)
|
||||
}
|
||||
return AskKey{Kind: kind, Node: node, Module: module, FP: fp}, nil
|
||||
case AskKindOutside:
|
||||
k, err := ParseAnyAskKey(AskKindRestore + "." + rest)
|
||||
if err != nil {
|
||||
return refused(fmt.Errorf("an outside ask is outside.<node>.<module>.<fingerprint hex>"))
|
||||
}
|
||||
k.Kind = kind
|
||||
return k, nil
|
||||
case AskKindBackup:
|
||||
node, module, part, isNode, err := nodeengineclient.ParseComposedKey(rest)
|
||||
if err != nil || isNode {
|
||||
|
||||
@@ -79,7 +79,7 @@ var readFamilies = []string{buckets.ComposedRead}
|
||||
var ownBuckets = []string{buckets.Asks, buckets.Differences,
|
||||
buckets.Puts, buckets.Wants, plc.AssignmentsBucket, plc.ProposalsBucket,
|
||||
plc.CandidatesBucket, plc.PulsesBucket, assign.FindingsBucket, assign.PoliciesBucket, plc.SeatHoldersBucket,
|
||||
engineReadsBucket, retiredBucket, copiesBucket, writtenBucket}
|
||||
engineReadsBucket, retiredBucket, copiesBucket, writtenBucket, buckets.Waits}
|
||||
|
||||
// engineReadsState is the assigner's state of what each node's engine states it reads (ADR 0618 §3), one key `<node>`, as
|
||||
// the verb manifest-reads last answered it or the fallback read it: what assigner_engine_reads lists and asks again.
|
||||
@@ -119,6 +119,7 @@ var keptStates = []announce.State{
|
||||
{Name: retiredState}, // read only by the copy into assignments (migrate.go)
|
||||
{Name: copiesState, Readable: true},
|
||||
{Name: writtenState, Readable: true},
|
||||
{Name: buckets.WaitsState, Readable: true},
|
||||
}
|
||||
|
||||
// kvSet is every bucket the assigner opened, by name.
|
||||
@@ -177,6 +178,12 @@ func open(ctx context.Context, js process.BucketOpener, logf func(string, ...any
|
||||
logf("the bucket %s cannot be opened (%v): what each node's engine reads is kept in memory only until it starts again", name, err)
|
||||
continue
|
||||
}
|
||||
if name == buckets.Waits && err != nil {
|
||||
// New with the wait on an outside cause (design 47 G §706): without it no wait is made, and a failing check is judged
|
||||
// as before.
|
||||
logf("the bucket %s cannot be opened (%v): no outside wait is kept until the assigner starts again", name, err)
|
||||
continue
|
||||
}
|
||||
if name == retiredBucket && err != nil {
|
||||
// The retired state is read only for the copy into `assignments` (migrate.go): a mesh made after the switch has
|
||||
// none, and nothing is copied.
|
||||
@@ -201,7 +208,7 @@ func open(ctx context.Context, js process.BucketOpener, logf func(string, ...any
|
||||
func startsWithout(name string) bool {
|
||||
switch name {
|
||||
case plc.SeatHoldersBucket, buckets.Provisions, assign.PoliciesBucket, assign.FindingsBucket,
|
||||
buckets.Puts, buckets.Source, buckets.Merges, engineReadsBucket, retiredBucket, writtenBucket:
|
||||
buckets.Puts, buckets.Source, buckets.Merges, engineReadsBucket, retiredBucket, writtenBucket, buckets.Waits:
|
||||
return true
|
||||
}
|
||||
return slices.Contains(seatBuckets, name) || name == buckets.Grants
|
||||
@@ -330,6 +337,7 @@ var watched = []string{
|
||||
buckets.Merges,
|
||||
engineReadsBucket,
|
||||
writtenBucket,
|
||||
buckets.Waits,
|
||||
}
|
||||
|
||||
// follow watches every bucket a pass reads and makes a pass at start, once every watch handed what stands, and after
|
||||
@@ -508,6 +516,27 @@ func follow(ctx context.Context, js jetstream.JetStream, nc process.Subscriber,
|
||||
}
|
||||
p := &passer{kvs: kvs, w: w, logf: logger.Printf, say: say, findings: findings, said: map[string]bool{}, answeredAt: map[string]uint64{},
|
||||
reads: reads, seats: gate}
|
||||
// The wait on an outside cause (outside.go, design 47 G §706): the keeper's three events say when to judge again, and the
|
||||
// seat mesh-wants' wait-held and wait-ended are said; registered before the hub starts.
|
||||
if bus != nil && kvs[buckets.Waits] != nil {
|
||||
bound, settle := outsideSettings(os.Getenv)
|
||||
p.outside = newOutsideActs(bus.CallKept, wake, logger.Printf, bound, settle)
|
||||
for _, subject := range []string{mcseat.SubjectConditionRaised, mcseat.SubjectConditionChanged, mcseat.SubjectConditionCleared} {
|
||||
subject := subject
|
||||
if err := hub.Handle(subject, func(m jetstream.Msg) {
|
||||
p.outside.Heard(m.Subject(), m.Data())
|
||||
if err := m.Ack(); err != nil {
|
||||
logger.Printf("acknowledging %s: %v", m.Subject(), err)
|
||||
}
|
||||
}); err != nil {
|
||||
logger.Printf("%s is not heard, so a wait on an outside cause is judged again only when its check changes: %v", subject, err)
|
||||
}
|
||||
}
|
||||
p.outside.readStart()
|
||||
}
|
||||
if js != nil {
|
||||
p.publishWants = wantsPublisher(ctx, js, thisNode)
|
||||
}
|
||||
// The node-engine seat's manifest-reads-changed, for one node (ADR 0618 §4): that node is asked again, with no timer.
|
||||
// Heard on the assigner's durable consumer on EVENTS, which the grants part makes from the manifest's consumes; where it
|
||||
// is not there yet, a node is asked again only when it comes back online.
|
||||
|
||||
@@ -0,0 +1,469 @@
|
||||
package main
|
||||
|
||||
import (
|
||||
"context"
|
||||
"encoding/json"
|
||||
"fmt"
|
||||
"strconv"
|
||||
"strings"
|
||||
"sync"
|
||||
"time"
|
||||
|
||||
"github.com/nats-io/nats.go"
|
||||
"github.com/nats-io/nats.go/jetstream"
|
||||
|
||||
"git.novox.be/novox/mesh-contracts/go/clients/nodeengineclient"
|
||||
"git.novox.be/novox/mesh-contracts/go/seats/nodeengine"
|
||||
mcseat "git.novox.be/novox/mesh-contracts/health-and-repair/go/seats/meshconditions"
|
||||
"git.novox.be/novox/mesh-delivery/contracts/go/clients/meshcatalogkey"
|
||||
meshwants "git.novox.be/novox/mesh-delivery/contracts/go/seats/meshwants"
|
||||
"git.novox.be/novox/mesh-delivery/modules/mesh-delivery-assigner/asks"
|
||||
"git.novox.be/novox/mesh-sdk/go/seatdef"
|
||||
"git.novox.be/novox/mesh-sdk/go/views"
|
||||
|
||||
"mesh-delivery-assigner/internal/assign"
|
||||
"mesh-delivery-assigner/internal/assignment"
|
||||
"mesh-delivery-assigner/internal/buckets"
|
||||
"mesh-delivery-assigner/internal/outside"
|
||||
)
|
||||
|
||||
// The wait on an outside cause (design 47 G §706, ADR 0677): the assigner judges a node's health after a want moves, and a moved
|
||||
// module's failing check on a machine whose open conditions name an outside cause is held as a wait (internal/outside has the
|
||||
// rules). The conditions come only through the seat mesh-conditions: its verb `conditions`, read at each judging as a kept
|
||||
// call, and its events, which only say when to judge again.
|
||||
|
||||
// conditionsVerb is the seat verb the keeper answers `conditions` by.
|
||||
const conditionsVerb = "seat:" + mcseat.Seat + "." + mcseat.VerbConditions
|
||||
|
||||
// wantsPublisher says one of the seat mesh-wants' events, `wait-held` and `wait-ended` among them, under a message id the bus
|
||||
// keeps one of.
|
||||
func wantsPublisher(ctx context.Context, js jetstream.JetStream, node string) publisher {
|
||||
return func(event string, body any, id string) error {
|
||||
data, err := json.Marshal(body)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
msg := nats.NewMsg(seatdef.EventSubject(meshwants.Seat, event))
|
||||
msg.Header.Set("x-event-id", id)
|
||||
msg.Header.Set("x-source", assignment.Module)
|
||||
msg.Header.Set("x-node", node)
|
||||
msg.Header.Set("x-time", time.Now().UTC().Format(time.RFC3339Nano))
|
||||
msg.Header.Set("content-type", "application/json")
|
||||
msg.Data = data
|
||||
_, err = js.PublishMsg(ctx, msg, jetstream.WithMsgID(id))
|
||||
return err
|
||||
}
|
||||
}
|
||||
|
||||
// outsideActs keeps, while the process runs, what the judging of outside waits needs beside the kept waits: the conditions read
|
||||
// for each machine (one answer is one judging), the machines a heard event says to judge again, the machines whose own network
|
||||
// is held open, and each check's last signature.
|
||||
type outsideActs struct {
|
||||
mu sync.Mutex
|
||||
call func(key string, body any) (json.RawMessage, error) // a kept call
|
||||
wake func()
|
||||
logf func(string, ...any)
|
||||
// bound and settle are the settings outside-wait-minutes and outside-settle-seconds.
|
||||
bound, settle time.Duration
|
||||
// reading is a read in flight by machine, answered the answers it brought, not yet judged by.
|
||||
reading map[string]bool
|
||||
answered map[string][]outside.Cond
|
||||
// dirty is a machine whose waits are judged again, cleared a wait's key whose cause's clearing was heard.
|
||||
dirty map[string]bool
|
||||
cleared map[string]bool
|
||||
// sigs is each check's last signature, by node.module.
|
||||
sigs map[string]string
|
||||
// own is the machines whose own network is open; ownRead says the read at the start was answered.
|
||||
own map[string]bool
|
||||
ownRead bool
|
||||
// causes is, by condition key, the nodes whose waits stand on it, as the last pass kept them.
|
||||
causes map[string][]string
|
||||
// started says the first pass judged every wait again.
|
||||
started bool
|
||||
}
|
||||
|
||||
func newOutsideActs(call func(string, any) (json.RawMessage, error), wake func(), logf func(string, ...any), bound, settle time.Duration) *outsideActs {
|
||||
if bound <= 0 {
|
||||
bound = outside.DefaultBound
|
||||
}
|
||||
if settle <= 0 {
|
||||
settle = outside.DefaultSettle
|
||||
}
|
||||
return &outsideActs{call: call, wake: wake, logf: logf, bound: bound, settle: settle, reading: map[string]bool{}, answered: map[string][]outside.Cond{},
|
||||
dirty: map[string]bool{}, cleared: map[string]bool{}, sigs: map[string]string{}, own: map[string]bool{}, causes: map[string][]string{}}
|
||||
}
|
||||
|
||||
// outsideSettings are the bounds of the wait, the settings outside-wait-minutes and outside-settle-seconds (design 47 G §633, §706),
|
||||
// from the environment the manifest's config fills; zero for one not set, which is its default.
|
||||
func outsideSettings(getenv func(string) string) (bound, settle time.Duration) {
|
||||
if n, err := strconv.Atoi(strings.TrimSpace(getenv("MESH_ASSIGNER_OUTSIDE_WAIT_MINUTES"))); err == nil && n > 0 {
|
||||
bound = time.Duration(n) * time.Minute
|
||||
}
|
||||
if n, err := strconv.Atoi(strings.TrimSpace(getenv("MESH_ASSIGNER_OUTSIDE_SETTLE_SECONDS"))); err == nil && n > 0 {
|
||||
settle = time.Duration(n) * time.Second
|
||||
}
|
||||
return bound, settle
|
||||
}
|
||||
|
||||
// readStart reads the open conditions of every machine once at the start, for the machines whose own network is held.
|
||||
func (o *outsideActs) readStart() {
|
||||
go func() {
|
||||
answer, err := o.call(conditionsVerb, map[string]any{"scope": "machine"})
|
||||
if err != nil {
|
||||
o.logf("the open conditions are not read at the start (%v): no machine is held for its own network until its event is heard", err)
|
||||
return
|
||||
}
|
||||
conds, err := outside.ParseConditions(answer)
|
||||
if err != nil {
|
||||
o.logf("%v", err)
|
||||
return
|
||||
}
|
||||
o.mu.Lock()
|
||||
for m := range outside.OwnNetwork(conds) {
|
||||
o.own[m] = true
|
||||
}
|
||||
o.ownRead = true
|
||||
o.mu.Unlock()
|
||||
o.wake()
|
||||
}()
|
||||
}
|
||||
|
||||
// Heard is one of the seat's events condition-raised, -changed or -cleared: it says which machines to judge again and keeps the
|
||||
// machines whose own network is open; nothing is kept of the condition.
|
||||
func (o *outsideActs) Heard(subject string, body []byte) {
|
||||
var e mcseat.ConditionEvent
|
||||
if err := json.Unmarshal(body, &e); err != nil || e.Key == "" {
|
||||
o.logf("%s holds what is not a condition's event, and is dropped: %v", subject, err)
|
||||
return
|
||||
}
|
||||
cleared := strings.HasSuffix(subject, "."+mcseat.EventConditionCleared)
|
||||
o.mu.Lock()
|
||||
if p := strings.Split(e.Key, "."); len(p) == 3 && p[0] == "machine" && p[2] == "network" {
|
||||
o.own[p[1]] = !cleared
|
||||
}
|
||||
if cleared {
|
||||
for _, node := range o.causes[e.Key] {
|
||||
o.dirty[node] = true
|
||||
o.cleared[node+"|"+e.Key] = true
|
||||
}
|
||||
} else if e.Subject != nil {
|
||||
o.dirty[e.Subject.Machine] = true
|
||||
for _, m := range e.Subject.Also {
|
||||
o.dirty[m] = true
|
||||
}
|
||||
}
|
||||
o.mu.Unlock()
|
||||
o.wake()
|
||||
}
|
||||
|
||||
// Held says whether machine's own network is open: no want moves to it.
|
||||
func (o *outsideActs) Held(machine string) bool {
|
||||
if o == nil {
|
||||
return false
|
||||
}
|
||||
o.mu.Lock()
|
||||
defer o.mu.Unlock()
|
||||
return o.own[machine]
|
||||
}
|
||||
|
||||
// take is the conditions read for machine, consumed (one answer is one judging), or false while the kept call is unanswered,
|
||||
// which it starts when none is in flight.
|
||||
func (o *outsideActs) take(machine string) ([]outside.Cond, bool) {
|
||||
o.mu.Lock()
|
||||
defer o.mu.Unlock()
|
||||
if c, ok := o.answered[machine]; ok {
|
||||
delete(o.answered, machine)
|
||||
return c, true
|
||||
}
|
||||
if o.reading[machine] {
|
||||
return nil, false
|
||||
}
|
||||
o.reading[machine] = true
|
||||
go func() {
|
||||
answer, err := o.call(conditionsVerb, map[string]any{"scope": "machine", "machine": machine})
|
||||
var conds []outside.Cond
|
||||
if err == nil {
|
||||
conds, err = outside.ParseConditions(answer)
|
||||
}
|
||||
o.mu.Lock()
|
||||
o.reading[machine] = false
|
||||
if err != nil {
|
||||
o.logf("the conditions of %s are not read (%v): the move waits on %s", machine, err, outside.OnUnread)
|
||||
} else {
|
||||
o.answered[machine] = conds
|
||||
o.dirty[machine] = true // the kept call answered: the move is judged by it
|
||||
}
|
||||
o.mu.Unlock()
|
||||
if err == nil {
|
||||
o.wake()
|
||||
}
|
||||
}()
|
||||
return nil, false
|
||||
}
|
||||
|
||||
// outsideReview is the judging of module's check at fp on each of its nodes (§706): it returns the nodes as the rules read them,
|
||||
// the have of a waiting move no failure (its health unknown) and a go-on's counted healthy, and keeps the waits, says their
|
||||
// events, finding and ask. moved says the delivery moved owner to node.
|
||||
func (p *passer) outsideReview(ctx context.Context, module, fp string, nodes []string, n assign.Nodes, keeperSelf bool, moved func(owner, node string) bool) assign.Nodes {
|
||||
o := p.outside
|
||||
if o == nil || fp == "" || p.kvs[buckets.Waits] == nil {
|
||||
return n
|
||||
}
|
||||
hex, err := meshcatalogkey.KeyPart(fp)
|
||||
if err != nil {
|
||||
return n
|
||||
}
|
||||
now := p.clock()
|
||||
kept := p.w.read(buckets.Waits)
|
||||
override := map[string]string{}
|
||||
// A wait of another fingerprint of this module is replaced by the newer want.
|
||||
for key, raw := range kept {
|
||||
if w, err := outside.Decode(raw); err == nil && w.Module == module && w.FP != fp {
|
||||
p.outsideEnd(ctx, key, w, outside.HowReplaced, now)
|
||||
}
|
||||
}
|
||||
for _, node := range nodes {
|
||||
key := outside.Key(node, module, hex)
|
||||
var prev *outside.Wait
|
||||
if raw := kept[key]; raw != nil {
|
||||
if w, err := outside.Decode(raw); err == nil {
|
||||
prev = &w
|
||||
}
|
||||
}
|
||||
h, ok := n.Haves[node][module]
|
||||
sig := fmt.Sprintf("%s|%s|%s", h.Fp, h.Phase, h.Health)
|
||||
o.mu.Lock()
|
||||
first := !o.started
|
||||
changed := o.sigs[node+"."+module] != sig
|
||||
o.sigs[node+"."+module] = sig
|
||||
dirty := o.dirty[node]
|
||||
cleared := prev != nil && prev.Cause != "" && o.cleared[node+"|"+prev.Cause]
|
||||
o.mu.Unlock()
|
||||
failing, healthy := ok && h.Fp == fp && assign.Failed(h), ok && assign.Healthy(h, fp)
|
||||
if !failing && !healthy && prev == nil {
|
||||
continue
|
||||
}
|
||||
if prev != nil && p.outsideAnswered(prev, node, module, fp) {
|
||||
dirty = true
|
||||
}
|
||||
due := prev != nil && (prev.Release != "" || (prev.Settling && !now.Before(prev.SettleEnds)))
|
||||
var d outside.Decision
|
||||
if prev != nil && !(first || changed || dirty || due) {
|
||||
// nothing to judge by: the wait stands as kept; the bound's ask is made when it has run
|
||||
d = outside.Decision{Wait: prev, Waiting: prev.Standing() || prev.Settling, Own: prev.Decided == outside.HowFailed || prev.Decided == outside.HowOwnFailure,
|
||||
Counted: prev.Decided == outside.HowGoOn}
|
||||
if prev.Standing() && !now.Before(prev.ClockFrom.Add(o.bound)) {
|
||||
d.Ask = true
|
||||
}
|
||||
} else {
|
||||
f := outside.Facts{Now: now, Failing: failing, Healthy: healthy, Node: node, KeeperSelf: keeperSelf, Bound: o.bound, Settle: o.settle,
|
||||
Moved: func(owner string) bool { return moved(owner, node) }, CauseCleared: cleared}
|
||||
if failing && !keeperSelf {
|
||||
f.Conds, f.Read = o.take(node)
|
||||
}
|
||||
d = outside.Judge(prev, node, module, fp, f)
|
||||
o.mu.Lock()
|
||||
delete(o.dirty, node)
|
||||
if prev != nil {
|
||||
delete(o.cleared, node+"|"+prev.Cause)
|
||||
}
|
||||
o.mu.Unlock()
|
||||
}
|
||||
switch {
|
||||
case d.Waiting:
|
||||
override[node] = nodeengineclient.HealthUnknown
|
||||
case d.Counted:
|
||||
override[node] = nodeengineclient.HealthOK
|
||||
}
|
||||
p.outsideApply(ctx, key, hex, node, module, fp, prev, d, now)
|
||||
}
|
||||
o.mu.Lock()
|
||||
o.started = true
|
||||
o.mu.Unlock()
|
||||
if len(override) == 0 {
|
||||
return n
|
||||
}
|
||||
out := n
|
||||
out.Haves = map[string]map[string]nodeengine.HaveByNodeModuleValue{}
|
||||
for node, byModule := range n.Haves {
|
||||
out.Haves[node] = map[string]nodeengine.HaveByNodeModuleValue{}
|
||||
for m, h := range byModule {
|
||||
out.Haves[node][m] = h
|
||||
}
|
||||
}
|
||||
for node, health := range override {
|
||||
h := out.Haves[node][module]
|
||||
h.Health = health
|
||||
if health == nodeengineclient.HealthOK {
|
||||
h.Phase = nodeengineclient.PhaseRunning
|
||||
}
|
||||
out.Haves[node][module] = h
|
||||
}
|
||||
return out
|
||||
}
|
||||
|
||||
// outsideAnswered reads the operator's answer to the wait's ask, from the teller's record of it under the ask's key (answeredWith,
|
||||
// §700's path as the assigner reads an ask's answer), and notes it on the wait as a release; it says whether it did.
|
||||
func (p *passer) outsideAnswered(w *outside.Wait, node, module, fp string) bool {
|
||||
if w.Ask == "" || w.Release != "" {
|
||||
return false
|
||||
}
|
||||
key, err := asks.OutsideAskKey(node, module, fp)
|
||||
if err != nil {
|
||||
return false
|
||||
}
|
||||
for _, how := range []string{assign.OptionGoOn, assign.OptionFailed} {
|
||||
if p.answeredWith(key, how) {
|
||||
w.Release, w.By, w.Why = how, "the operator", "the answer to the ask "+w.Ask
|
||||
return true
|
||||
}
|
||||
}
|
||||
return false
|
||||
}
|
||||
|
||||
// outsideApply keeps what a judging came to: the wait's record, its events, its finding and its ask.
|
||||
func (p *passer) outsideApply(ctx context.Context, key, hex, node, module, fp string, prev *outside.Wait, d outside.Decision, now time.Time) {
|
||||
o := p.outside
|
||||
switch {
|
||||
case d.End != "":
|
||||
p.outsideEnd(ctx, key, *prev, d.End, now)
|
||||
if d.Wait != nil {
|
||||
p.keepWait(ctx, key, *d.Wait)
|
||||
}
|
||||
return
|
||||
case d.Wait == nil:
|
||||
if prev != nil {
|
||||
p.outsideEnd(ctx, key, *prev, "", now)
|
||||
}
|
||||
return
|
||||
}
|
||||
w := *d.Wait
|
||||
if d.Held {
|
||||
body := p.outsideBody(w)
|
||||
if err := p.wantsSay(meshwants.EventWaitHeld, body, key+"."+w.On+"."+w.ClockFrom.UTC().Format(time.RFC3339Nano)); err != nil {
|
||||
p.logf("the wait of %s on %s is not said, and is said at the next pass: %v", module, node, err)
|
||||
return
|
||||
}
|
||||
p.outsideFinding(w, true)
|
||||
}
|
||||
if d.Ask && w.Ask == "" {
|
||||
if id := p.outsideAsk(ctx, node, module, fp, w, now); id != "" {
|
||||
w.Ask = id
|
||||
}
|
||||
}
|
||||
if w.Standing() || w.Settling || w.Decided != "" {
|
||||
o.mu.Lock()
|
||||
if w.Cause != "" && w.Standing() {
|
||||
if !containsString(o.causes[w.Cause], node) {
|
||||
o.causes[w.Cause] = append(o.causes[w.Cause], node)
|
||||
}
|
||||
}
|
||||
o.mu.Unlock()
|
||||
if w.Settling {
|
||||
p.wakeAt(w.SettleEnds, now)
|
||||
} else if w.Standing() {
|
||||
p.wakeAt(w.ClockFrom.Add(o.bound), now)
|
||||
}
|
||||
p.keepWait(ctx, key, w)
|
||||
}
|
||||
}
|
||||
|
||||
func containsString(l []string, s string) bool {
|
||||
for _, x := range l {
|
||||
if x == s {
|
||||
return true
|
||||
}
|
||||
}
|
||||
return false
|
||||
}
|
||||
|
||||
// keepWait writes the wait's record.
|
||||
func (p *passer) keepWait(ctx context.Context, key string, w outside.Wait) {
|
||||
p.put(ctx, buckets.Waits, key, func() ([]byte, error) { return outside.Encode(w) })
|
||||
}
|
||||
|
||||
func (p *passer) outsideBody(w outside.Wait) meshwants.Wait {
|
||||
since, from := w.Since, w.ClockFrom
|
||||
return meshwants.Wait{Step: meshwants.WaitStepOutside, Owner: assignment.Module, Node: w.Node, Module: w.Module, Fingerprint: w.FP, On: w.On,
|
||||
Since: since, ClockFrom: &from, Bound: w.Bound,
|
||||
Words: fmt.Sprintf("%s on %s waits on %s", w.Module, w.Node, w.On)}
|
||||
}
|
||||
|
||||
// outsideEnd ends a wait: its wait-ended with how (none for a settle, which was never said), its finding and ask taken away, and
|
||||
// its record deleted unless the end decided it (the caller keeps that record).
|
||||
func (p *passer) outsideEnd(ctx context.Context, key string, w outside.Wait, how string, now time.Time) {
|
||||
if w.Standing() && how != "" {
|
||||
body := p.outsideBody(w)
|
||||
body.How = how
|
||||
if err := p.wantsSay(meshwants.EventWaitEnded, body, key+"."+how+"."+now.UTC().Format(time.RFC3339Nano)); err != nil {
|
||||
p.logf("the end of the wait of %s on %s is not said, and is said at the next pass: %v", w.Module, w.Node, err)
|
||||
return
|
||||
}
|
||||
}
|
||||
p.outsideFinding(w, false)
|
||||
if ask, err := asks.OutsideAskKey(w.Node, w.Module, w.FP); err == nil {
|
||||
p.remove(ctx, buckets.Asks, ask)
|
||||
}
|
||||
if p.outside != nil && w.Cause != "" {
|
||||
p.outside.mu.Lock()
|
||||
p.outside.causes[w.Cause] = removeString(p.outside.causes[w.Cause], w.Node)
|
||||
p.outside.mu.Unlock()
|
||||
}
|
||||
p.remove(ctx, buckets.Waits, key)
|
||||
}
|
||||
|
||||
func removeString(l []string, s string) []string {
|
||||
var out []string
|
||||
for _, x := range l {
|
||||
if x != s {
|
||||
out = append(out, x)
|
||||
}
|
||||
}
|
||||
return out
|
||||
}
|
||||
|
||||
// outsideFinding writes the finding delivery-waits-outside, or clears it.
|
||||
func (p *passer) outsideFinding(w outside.Wait, raised bool) {
|
||||
if p.findings.Put == nil {
|
||||
return
|
||||
}
|
||||
o := assign.WaitsOutsideFinding(w.Node, w.Module, w.Cause, w.On, p.clock())
|
||||
if raised {
|
||||
if err := p.findings.Put(o.Finding(p.clock())); err != nil {
|
||||
p.logf("the finding %s is not written: %v", "delivery."+o.ID, err)
|
||||
}
|
||||
return
|
||||
}
|
||||
if p.findings.Delete != nil {
|
||||
if err := p.findings.Delete(mcseatKey(o)); err != nil {
|
||||
p.logf("the finding %s is not cleared: %v", "delivery."+o.ID, err)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
func mcseatKey(o assign.Observed) string {
|
||||
return views.FindingKey(o.Scope, assign.KeyID(o.ID), o.Kind)
|
||||
}
|
||||
|
||||
// outsideAsk asks the operator at the wait's bound, once, and returns the asking's id.
|
||||
func (p *passer) outsideAsk(ctx context.Context, node, module, fp string, w outside.Wait, now time.Time) string {
|
||||
key, err := asks.OutsideAskKey(node, module, fp)
|
||||
if err != nil {
|
||||
return ""
|
||||
}
|
||||
asking := assign.AskingSeconds(now)
|
||||
a := assign.OutsideAsk(key, asks.OutsideAbout(key), node, module, fp, w.On, w.ClockFrom, asking, now)
|
||||
p.put(ctx, buckets.Asks, key, func() ([]byte, error) { return assign.EncodeAsk(key, a) })
|
||||
if ask, err := assign.AskOf(a); err == nil {
|
||||
return ask.ID
|
||||
}
|
||||
return ""
|
||||
}
|
||||
|
||||
// wakeAt makes a pass when t comes: the settle's end and the bound are the waits' own times, the one wake (no polling).
|
||||
func (p *passer) wakeAt(t, now time.Time) {
|
||||
if p.outside == nil || p.outside.wake == nil {
|
||||
return
|
||||
}
|
||||
time.AfterFunc(max(t.Sub(now), 0)+time.Millisecond, p.outside.wake)
|
||||
}
|
||||
@@ -0,0 +1,273 @@
|
||||
package main
|
||||
|
||||
import (
|
||||
"context"
|
||||
"encoding/json"
|
||||
"strings"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
"git.novox.be/novox/mesh-contracts/go/clients/nodeengineclient"
|
||||
"git.novox.be/novox/mesh-contracts/go/seats/nodeengine"
|
||||
mcseat "git.novox.be/novox/mesh-contracts/health-and-repair/go/seats/meshconditions"
|
||||
mdc "git.novox.be/novox/mesh-delivery/contracts/go/clients/meshdeliveryclient"
|
||||
meshwants "git.novox.be/novox/mesh-delivery/contracts/go/seats/meshwants"
|
||||
"git.novox.be/novox/mesh-delivery/modules/mesh-delivery-assigner/asks"
|
||||
|
||||
"mesh-delivery-assigner/internal/assign"
|
||||
"mesh-delivery-assigner/internal/buckets"
|
||||
"mesh-delivery-assigner/internal/outside"
|
||||
)
|
||||
|
||||
// The whole of the wait on an outside cause in the pass (design 47 G §706): a fixture keeper answering `conditions` and saying
|
||||
// its events, a failing check of nats on ace, and the events, finding, ask and record the pass leaves.
|
||||
type outsideRig struct {
|
||||
t *testing.T
|
||||
p *passer
|
||||
now time.Time
|
||||
conds []map[string]any
|
||||
unread bool
|
||||
events []string
|
||||
bodies []meshwants.Wait
|
||||
found map[string]bool
|
||||
woke chan struct{}
|
||||
health string
|
||||
}
|
||||
|
||||
func newOutsideRig(t *testing.T) *outsideRig {
|
||||
r := &outsideRig{t: t, now: time.Date(2026, 11, 11, 10, 0, 0, 0, time.UTC), found: map[string]bool{}, woke: make(chan struct{}, 16), health: nodeengineclient.HealthFailed}
|
||||
r.p = newPasser(kvSet{buckets.Waits: &memKV{values: map[string][]byte{}}, buckets.Asks: &memKV{values: map[string][]byte{}}}, r.now)
|
||||
r.p.now = func() time.Time { return r.now }
|
||||
r.p.w.put(buckets.Asks, "x", nil)
|
||||
r.p.outside = newOutsideActs(func(key string, body any) (json.RawMessage, error) {
|
||||
if r.unread {
|
||||
return nil, context.DeadlineExceeded
|
||||
}
|
||||
return json.Marshal(map[string]any{"conditions": r.conds})
|
||||
}, func() { r.woke <- struct{}{} }, func(string, ...any) {}, 30*time.Minute, 90*time.Second)
|
||||
r.p.publishWants = func(event string, body any, id string) error {
|
||||
r.events = append(r.events, event)
|
||||
r.bodies = append(r.bodies, body.(meshwants.Wait))
|
||||
return nil
|
||||
}
|
||||
r.p.findings.Put = func(f mcseat.Finding) error { r.found[f.Scope+"."+f.ID+"."+f.Kind] = true; return nil }
|
||||
r.p.findings.Delete = func(key string) error { delete(r.found, key); return nil }
|
||||
return r
|
||||
}
|
||||
|
||||
func (r *outsideRig) cond(key string, also ...string) {
|
||||
r.conds = append(r.conds, map[string]any{"key": key, "subject": map[string]any{"machine": strings.Split(key, ".")[1], "also": also}})
|
||||
}
|
||||
|
||||
// review is one pass's judging of nats on ace at the fingerprint, with the read answered at once when one is started.
|
||||
func (r *outsideRig) review(keeperSelf bool, moved func(string, string) bool) assign.Nodes {
|
||||
r.t.Helper()
|
||||
n := assign.Nodes{Online: map[string]bool{"ace": true}, Haves: map[string]map[string]nodeengine.HaveByNodeModuleValue{
|
||||
"ace": {"nats": {Fp: fpTest, Phase: nodeengineclient.PhaseRunning, Health: r.health}}}}
|
||||
if moved == nil {
|
||||
moved = func(string, string) bool { return false }
|
||||
}
|
||||
out := r.p.outsideReview(context.Background(), "nats", fpTest, []string{"ace"}, n, keeperSelf, moved)
|
||||
select {
|
||||
case <-r.woke: // the read was answered: the next pass judges by it
|
||||
out = r.p.outsideReview(context.Background(), "nats", fpTest, []string{"ace"}, n, keeperSelf, moved)
|
||||
case <-time.After(50 * time.Millisecond):
|
||||
}
|
||||
return out
|
||||
}
|
||||
|
||||
func (r *outsideRig) wait() outside.Wait {
|
||||
r.t.Helper()
|
||||
for _, v := range r.p.w.read(buckets.Waits) {
|
||||
w, err := outside.Decode(v)
|
||||
if err != nil {
|
||||
r.t.Fatal(err)
|
||||
}
|
||||
return w
|
||||
}
|
||||
return outside.Wait{}
|
||||
}
|
||||
|
||||
func TestTheWaitOnAnOutsideCause(t *testing.T) {
|
||||
healthOf := func(n assign.Nodes) string { return n.Haves["ace"]["nats"].Health }
|
||||
t.Run("a cause listing the machine is a wait said at once, ended cleared when it clears healthy", func(t *testing.T) {
|
||||
r := newOutsideRig(t)
|
||||
r.cond("machine.hub.unreachable", "ace")
|
||||
n := r.review(false, nil)
|
||||
if healthOf(n) == nodeengineclient.HealthFailed || len(r.events) != 1 || r.events[0] != meshwants.EventWaitHeld {
|
||||
t.Fatalf("health %s events %v", healthOf(n), r.events)
|
||||
}
|
||||
b := r.bodies[0]
|
||||
if b.Step != meshwants.WaitStepOutside || b.On != "unreachable:hub" || b.Bound != "30m0s" || b.ClockFrom == nil || !b.ClockFrom.Equal(r.now) || b.Node != "ace" || b.Module != "nats" {
|
||||
t.Errorf("wait-held %+v", b)
|
||||
}
|
||||
if !r.found["delivery.ace.nats.delivery-waits-outside"] {
|
||||
t.Errorf("findings %v", r.found)
|
||||
}
|
||||
// the cause clears and the module runs healthy
|
||||
r.conds, r.health, r.now = nil, nodeengineclient.HealthOK, r.now.Add(time.Minute)
|
||||
r.p.outside.Heard("mesh.seat.mesh-conditions.event.condition-cleared", []byte(`{"key":"machine.hub.unreachable","event":"cleared","change":"cleared","at":"2026-11-11T10:01:00Z","subject":{"id":"hub","scope":"machine","machine":"hub","also":["ace"]}}`))
|
||||
<-r.woke
|
||||
r.review(false, nil)
|
||||
if len(r.events) != 2 || r.events[1] != meshwants.EventWaitEnded || r.bodies[1].How != outside.HowCleared || len(r.found) != 0 || len(r.p.w.read(buckets.Waits)) != 0 {
|
||||
t.Errorf("events %v bodies %+v findings %v", r.events, r.bodies, r.found)
|
||||
}
|
||||
})
|
||||
t.Run("a cause not listing the machine settles, then the failure is the delivery's", func(t *testing.T) {
|
||||
r := newOutsideRig(t)
|
||||
r.cond("machine.hub.unreachable", "g14")
|
||||
n := r.review(false, nil)
|
||||
if healthOf(n) == nodeengineclient.HealthFailed || len(r.events) != 0 || !r.wait().Settling {
|
||||
t.Fatalf("settle: health %s events %v wait %+v", healthOf(n), r.events, r.wait())
|
||||
}
|
||||
r.now = r.now.Add(91 * time.Second)
|
||||
if n := r.review(false, nil); healthOf(n) != nodeengineclient.HealthFailed {
|
||||
t.Errorf("past the settle the check still counts as failing: %s", healthOf(n))
|
||||
}
|
||||
})
|
||||
t.Run("a cause turning into another keeps the wait and its clock; a rewritten naming a moved module fails at once", func(t *testing.T) {
|
||||
r := newOutsideRig(t)
|
||||
r.cond("machine.hub.unreachable", "ace")
|
||||
r.review(false, nil)
|
||||
r.conds = nil
|
||||
r.cond("machine.hub.network", "ace")
|
||||
r.now = r.now.Add(5 * time.Minute)
|
||||
r.p.outside.Heard("mesh.seat.mesh-conditions.event.condition-raised", []byte(`{"key":"machine.hub.network","subject":{"id":"hub","scope":"machine","machine":"hub","also":["ace"]}}`))
|
||||
<-r.woke
|
||||
r.review(false, nil)
|
||||
last := len(r.bodies) - 1
|
||||
if len(r.events) != 2 || r.events[last] != meshwants.EventWaitHeld || r.bodies[last].On != "network:hub" || !r.bodies[last].ClockFrom.Equal(*r.bodies[0].ClockFrom) {
|
||||
t.Errorf("events %v bodies %+v", r.events, r.bodies)
|
||||
}
|
||||
r2 := newOutsideRig(t)
|
||||
r2.cond("machine.ace.nats.rewritten")
|
||||
n := r2.review(false, func(owner, node string) bool { return owner == "nats" })
|
||||
if healthOf(n) == nodeengineclient.HealthFailed || !r2.wait().Settling {
|
||||
t.Errorf("a rewritten of the moved module is its failure: %s %+v", healthOf(n), r2.wait())
|
||||
}
|
||||
})
|
||||
t.Run("the keeper silent waits on conditions-unread, and its own module is judged with no read", func(t *testing.T) {
|
||||
r := newOutsideRig(t)
|
||||
r.unread = true
|
||||
r.review(false, nil)
|
||||
r.now = r.now.Add(90 * time.Second) // the settle has run with no answer
|
||||
r.review(false, nil)
|
||||
if len(r.bodies) != 1 || r.bodies[0].On != outside.OnUnread {
|
||||
t.Fatalf("bodies %+v", r.bodies)
|
||||
}
|
||||
r2 := newOutsideRig(t)
|
||||
r2.cond("machine.hub.unreachable", "ace")
|
||||
if n := r2.review(true, nil); healthOf(n) != nodeengineclient.HealthFailed || len(r2.events) != 0 {
|
||||
t.Errorf("the keeper's own failing check waits: %s %v", healthOf(n), r2.events)
|
||||
}
|
||||
})
|
||||
t.Run("past the bound the operator is asked; Go on counts it healthy and Count as failed holds it, each said ended", func(t *testing.T) {
|
||||
for how, want := range map[string]string{assign.OptionGoOn: nodeengineclient.HealthOK, assign.OptionFailed: nodeengineclient.HealthFailed} {
|
||||
r := newOutsideRig(t)
|
||||
r.cond("machine.hub.silent", "ace")
|
||||
r.review(false, nil)
|
||||
r.now = r.now.Add(31 * time.Minute)
|
||||
r.p.outside.Heard("mesh.seat.mesh-conditions.event.condition-changed", []byte(`{"key":"machine.hub.silent","subject":{"id":"hub","scope":"machine","machine":"hub","also":["ace"]}}`))
|
||||
<-r.woke
|
||||
r.review(false, nil)
|
||||
key, _ := asks.OutsideAskKey("ace", "nats", fpTest)
|
||||
raw := r.p.w.value(buckets.Asks, key)
|
||||
if raw == nil {
|
||||
t.Fatalf("%s: no ask at %s", how, key)
|
||||
}
|
||||
_, ask, err := assign.DecodeAsk(key, raw)
|
||||
if err != nil || ask.About != "ace.nats."+strings.TrimPrefix(key[strings.LastIndex(key, ".")+1:], "") || len(ask.Options) != 2 {
|
||||
t.Fatalf("ask %+v %v", ask, err)
|
||||
}
|
||||
v, _ := mdc.EncodeAnswer(mdc.Answer{Option: how, Warrant: ask.ID + "@2026-11-11T10:32:00Z", Words: "the operator"})
|
||||
r.p.w.put(buckets.Answers, key, v)
|
||||
n := r.review(false, nil)
|
||||
if r.events[len(r.events)-1] != meshwants.EventWaitEnded || r.bodies[len(r.bodies)-1].How != how || healthOf(n) != want {
|
||||
t.Errorf("%s: events %v last %+v health %s", how, r.events, r.bodies[len(r.bodies)-1], healthOf(n))
|
||||
}
|
||||
if r.p.w.value(buckets.Asks, key) != nil || len(r.found) != 0 {
|
||||
t.Errorf("%s: the ask or the finding stands after the end", how)
|
||||
}
|
||||
// a check that goes on failing is not waited on again
|
||||
r.now = r.now.Add(time.Minute)
|
||||
r.p.outside.Heard("mesh.seat.mesh-conditions.event.condition-raised", []byte(`{"key":"machine.hub.silent","subject":{"id":"hub","scope":"machine","machine":"hub","also":["ace"]}}`))
|
||||
<-r.woke
|
||||
before := len(r.events)
|
||||
r.review(false, nil)
|
||||
if len(r.events) != before {
|
||||
t.Errorf("%s: a decided check is waited on again", how)
|
||||
}
|
||||
}
|
||||
})
|
||||
t.Run("a newer want ends the wait replaced, and a restart judges it again", func(t *testing.T) {
|
||||
r := newOutsideRig(t)
|
||||
r.cond("machine.hub.unreachable", "ace")
|
||||
r.review(false, nil)
|
||||
n := assign.Nodes{Online: map[string]bool{"ace": true}, Haves: map[string]map[string]nodeengine.HaveByNodeModuleValue{}}
|
||||
r.p.outsideReview(context.Background(), "nats", "src2:"+strings.Repeat("c", 64), []string{"ace"}, n, false, func(string, string) bool { return false })
|
||||
if r.events[len(r.events)-1] != meshwants.EventWaitEnded || r.bodies[len(r.bodies)-1].How != outside.HowReplaced {
|
||||
t.Errorf("events %v", r.events)
|
||||
}
|
||||
})
|
||||
}
|
||||
|
||||
func TestAMachineWhoseOwnNetworkIsOpenGetsNoWant(t *testing.T) {
|
||||
m := assign.Module{Name: "nats", FP: fpTest, Built: true, Nodes: []string{"ace"}, MachineHeld: func(node string) bool { return node == "ace" }}
|
||||
ok, rule, why := assign.Holds(m, "ace", assign.Nodes{Online: map[string]bool{"ace": true}})
|
||||
if ok || rule == 0 || !strings.Contains(why, "machine.ace.network") {
|
||||
t.Errorf("holds = %v %d %q", ok, rule, why)
|
||||
}
|
||||
o := newOutsideActs(nil, func() {}, func(string, ...any) {}, 0, 0)
|
||||
o.Heard("mesh.seat.mesh-conditions.event.condition-raised", []byte(`{"key":"machine.ace.network","subject":{"id":"ace","scope":"machine","machine":"ace"}}`))
|
||||
if !o.Held("ace") || o.Held("g14") {
|
||||
t.Error("the own network condition's event does not hold the machine")
|
||||
}
|
||||
o.Heard("mesh.seat.mesh-conditions.event.condition-cleared", []byte(`{"key":"machine.ace.network","subject":{"id":"ace","scope":"machine","machine":"ace"}}`))
|
||||
if o.Held("ace") {
|
||||
t.Error("the machine is held after its condition cleared")
|
||||
}
|
||||
}
|
||||
|
||||
func TestTheToolsReadAndReleaseAnOutsideWait(t *testing.T) {
|
||||
r := newOutsideRig(t)
|
||||
r.cond("machine.hub.unreachable", "ace")
|
||||
r.review(false, nil)
|
||||
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
|
||||
})
|
||||
for k, v := range r.p.w.read(buckets.Waits) {
|
||||
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 {
|
||||
t.Fatalf("assigner_waits = %+v %v", got, err)
|
||||
}
|
||||
if w := got.(map[string]any)["waits"].([]waitSaid)[0]; 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 {
|
||||
t.Error("an unknown how is taken")
|
||||
}
|
||||
if _, err := s.doRelease(map[string]any{"node": "ace", "module": "nats", "how": "go-on"}, "shanks.node-tools"); err == nil || len(acts) != 0 {
|
||||
t.Error("a release with no why is taken")
|
||||
}
|
||||
if _, err := s.doRelease(map[string]any{"node": "ace", "module": "nats", "how": "go-on", "why": "the hub is being rebuilt"}, "shanks.node-tools"); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if len(acts) == 0 || !strings.Contains(acts[0], "hand-act") {
|
||||
t.Errorf("the hand act is not recorded first: %v", acts)
|
||||
}
|
||||
keys, _ := s.st.keys(waitsState)
|
||||
raw, _, _, _ := s.st.get(waitsState, keys[0])
|
||||
if w, _ := outside.Decode(raw); w.Release != "go-on" || !strings.Contains(w.By, "the hub is being rebuilt") {
|
||||
t.Errorf("wait = %+v", w)
|
||||
}
|
||||
// the process takes the release at its next pass: ended go-on, counted healthy
|
||||
r.p.w.put(buckets.Waits, keys[0], raw)
|
||||
n := r.review(false, nil)
|
||||
if r.bodies[len(r.bodies)-1].How != "go-on" || n.Haves["ace"]["nats"].Health != nodeengineclient.HealthOK {
|
||||
t.Errorf("last %+v health %s", r.bodies[len(r.bodies)-1], n.Haves["ace"]["nats"].Health)
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,120 @@
|
||||
package main
|
||||
|
||||
import (
|
||||
"errors"
|
||||
"fmt"
|
||||
"slices"
|
||||
"sort"
|
||||
"strings"
|
||||
"time"
|
||||
|
||||
"git.novox.be/novox/mesh-contracts/health-and-repair/go/clients/meshhandactsclient"
|
||||
stdio "git.novox.be/novox/mesh-sdk/go"
|
||||
|
||||
"mesh-delivery-assigner/internal/buckets"
|
||||
"mesh-delivery-assigner/internal/outside"
|
||||
)
|
||||
|
||||
// The tools of the waits on an outside cause (design 47 G §706, §276's row *delivery on an outside cause*): `assigner_waits` reads
|
||||
// each, the settle included, and `assigner_release` with `how` takes one. The tool writes the release into the wait's record and the
|
||||
// assigner's process ends the wait at its next pass, saying wait-ended with that how.
|
||||
|
||||
// toolWaits is the tool that reads the outside waits.
|
||||
const toolWaits = "assigner_waits"
|
||||
|
||||
// waitsState is the assigner's state of waits, as the runtime names a module's own state.
|
||||
const waitsState = buckets.WaitsState
|
||||
|
||||
// waitSaid is one wait as assigner_waits answers it: its record and where it stands.
|
||||
type waitSaid struct {
|
||||
Key string `json:"key"`
|
||||
outside.Wait
|
||||
// Settling's end is in the record; Ends is when the bound runs from clock-from, empty for a settle.
|
||||
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)
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("the waits cannot be listed: %w", err)
|
||||
}
|
||||
sort.Strings(keys)
|
||||
out := []waitSaid{}
|
||||
for _, key := range keys {
|
||||
raw, _, found, err := s.st.get(waitsState, key)
|
||||
if err != nil || !found {
|
||||
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
|
||||
}
|
||||
said := waitSaid{Key: key, Wait: w}
|
||||
if w.Standing() {
|
||||
if d, err := time.ParseDuration(w.Bound); err == nil {
|
||||
said.BoundEnds = w.ClockFrom.Add(d).UTC().Format("2006-01-02T15:04:05Z")
|
||||
}
|
||||
}
|
||||
out = append(out, said)
|
||||
}
|
||||
return map[string]any{"waits": out}, nil
|
||||
}
|
||||
|
||||
// doReleaseWait is assigner_release with how: the hand act is recorded first, then the release is written on the wait of node and
|
||||
// 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 !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)
|
||||
}
|
||||
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)
|
||||
}
|
||||
prefix := node + "." + module + "."
|
||||
var key string
|
||||
var w outside.Wait
|
||||
for _, k := range keys {
|
||||
if !strings.HasPrefix(k, prefix) {
|
||||
continue
|
||||
}
|
||||
raw, _, found, err := s.st.get(waitsState, k)
|
||||
if err != nil || !found {
|
||||
continue
|
||||
}
|
||||
if c, err := outside.Decode(raw); err == nil && c.Decided == "" {
|
||||
key, w = k, c
|
||||
}
|
||||
}
|
||||
if key == "" {
|
||||
return nil, fmt.Errorf("%s holds no wait of %s on an outside cause: 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, how), Why: why,
|
||||
Cause: "outside-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)
|
||||
}
|
||||
w.Release, w.By, w.Why = how, by+": "+why, why
|
||||
data, err := outside.Encode(w)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
if err := s.st.put(waitsState, key, data); err != nil {
|
||||
return nil, fmt.Errorf("the release of %s on %s is not written: %w", module, node, err)
|
||||
}
|
||||
return map[string]any{"released": key, "how": how, "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 " +
|
||||
"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.",
|
||||
Input: schema(map[string]string{"node": "one machine", "module": "one module"}, nil), Run: s.doWaits}
|
||||
}
|
||||
@@ -29,6 +29,7 @@ import (
|
||||
|
||||
"git.novox.be/novox/mesh-contracts/go/clients/nodeengineclient/enginereads"
|
||||
mcc "git.novox.be/novox/mesh-contracts/health-and-repair/go/clients/meshconditionsclient"
|
||||
mcseat "git.novox.be/novox/mesh-contracts/health-and-repair/go/seats/meshconditions"
|
||||
|
||||
"mesh-delivery-assigner/internal/assign"
|
||||
plc "mesh-delivery-assigner/internal/assignment"
|
||||
@@ -96,6 +97,18 @@ type passer struct {
|
||||
// each node's record, as the rules did before the verb. online is, by node, whether the last pass saw it online.
|
||||
reads *enginereads.Keeper
|
||||
online map[string]bool
|
||||
// outside is the judging of waits on an outside cause (outside.go, design 47 G §706); publishWants says the seat mesh-wants'
|
||||
// wait-held and wait-ended. Nil: no wait is made.
|
||||
outside *outsideActs
|
||||
publishWants publisher
|
||||
}
|
||||
|
||||
// wantsSay says one event of the seat mesh-wants; with no bus to say it on, nothing is said.
|
||||
func (p *passer) wantsSay(event string, body any, id string) error {
|
||||
if p.publishWants == nil {
|
||||
return nil
|
||||
}
|
||||
return p.publishWants(event, body, id)
|
||||
}
|
||||
|
||||
// engineReads sets what the rules read of each node's engine (ADR 0618 §3, §4): the statement the verb manifest-reads
|
||||
@@ -329,6 +342,10 @@ func (p *passer) wants(ctx context.Context) {
|
||||
for _, node := range m.Nodes {
|
||||
last[node] = p.w.lastHealthy(node, name, fp)
|
||||
}
|
||||
if p.outside != nil {
|
||||
m.MachineHeld = p.outside.Held
|
||||
nodes = p.outsideReview(ctx, name, fp, m.Nodes, nodes, manifests[name].ClaimsSeat(mcseat.Seat), p.movedTo(wanted, nodes))
|
||||
}
|
||||
out := assign.Wants(m, nodes, wantedOf[name], last)
|
||||
for _, node := range m.Nodes {
|
||||
for _, prov := range m.MissingProvisions(node) {
|
||||
@@ -384,6 +401,15 @@ func (p *passer) wants(ctx context.Context) {
|
||||
}
|
||||
}
|
||||
|
||||
// movedTo says whether the delivery moved owner to node: owner's want there names a fingerprint its have there does not yet run
|
||||
// healthy (design 47 G §706, *What counts as an outside cause*: a rewritten file of a module the delivery moved is that module's).
|
||||
func (p *passer) movedTo(wanted map[string]map[string]mwc.Want, n assign.Nodes) func(owner, node string) bool {
|
||||
return func(owner, node string) bool {
|
||||
w, ok := wanted[owner][node]
|
||||
return ok && !w.Removed && !assign.Healthy(n.Haves[node][owner], w.Fp)
|
||||
}
|
||||
}
|
||||
|
||||
// sayNoWant says in the journal why a node of module gets no want after this pass, from the rule the outcome names for it
|
||||
// (issue 1945: a new assignment whose trunk build was not done was held back with nothing said), once per reason, and says
|
||||
// when the node has a want again. A node with a live want, written before and kept or written in this pass, has one.
|
||||
|
||||
@@ -214,6 +214,9 @@ func followPut(node, module string) string {
|
||||
|
||||
// doRelease is the tool assigner_release.
|
||||
func (s *putTools) doRelease(args map[string]any, by string) (any, error) {
|
||||
if arg(args, "how") != "" {
|
||||
return s.doReleaseWait(args, by) // a wait on an outside cause (outside_tools.go, design 47 G §706)
|
||||
}
|
||||
node, module, why := arg(args, "node"), arg(args, "module"), arg(args, "why")
|
||||
key, err := puts.Key(node, module)
|
||||
if err != nil {
|
||||
@@ -344,9 +347,12 @@ func (s *putTools) tools() []stdio.Tool {
|
||||
Run: func(a map[string]any) (any, error) { return s.doPut(a, stdio.CallerOf(a)) }},
|
||||
{Name: toolRelease, Description: "End a put on one machine (novox/hq ADR 0546): its hold is taken away and the assigner's " +
|
||||
"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"},
|
||||
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"},
|
||||
[]string{"node", "module", "why"}),
|
||||
Run: func(a map[string]any) (any, error) { return s.doRelease(a, stdio.CallerOf(a)) }},
|
||||
s.waitsTool(),
|
||||
{Name: toolPuts, Description: "Every put and where it stands (novox/hq ADR 0546): its version, hold, why, each step with " +
|
||||
"its time and words. Narrowed by machine or module. Read-only.",
|
||||
Input: schema(map[string]string{"node": "one machine", "module": "one module"}, nil),
|
||||
|
||||
@@ -66,6 +66,46 @@ func RestoreAsk(key, node, module, fp, words string, asking int, now time.Time)
|
||||
})
|
||||
}
|
||||
|
||||
// The options of the outside ask (design 47 G §706).
|
||||
const (
|
||||
OptionGoOn = "go-on"
|
||||
OptionFailed = "failed"
|
||||
)
|
||||
|
||||
// OutsideAsk is the ask at a wait's bound (design 47 G §706): the move of module on node waits on cause since since, and the
|
||||
// operator says to go on (the module counts healthy there by the operator's word) or to count it as failed (held as the canary
|
||||
// rule has it). about is the ask's About, `<node>.<module>.<fingerprint hex>`.
|
||||
func OutsideAsk(key, about, node, module, fp, cause string, since time.Time, asking int, now time.Time) asks.Ask {
|
||||
return assignerAsk(asking, oc.Ask{
|
||||
ID: meshdeliveryclient.AskingID(key, asking),
|
||||
Headline: fmt.Sprintf("%s on %s waits on %s", module, node, cause),
|
||||
Explanation: fmt.Sprintf("The health check of %s at %s fails on %s, and the delivery has waited since %s on %s, which is outside it. "+
|
||||
"Go on counts %s healthy there by your word and the delivery goes on; count it as failed holds it as a failing check is held.",
|
||||
module, short(fp), node, since.UTC().Format(time.RFC3339), cause, module),
|
||||
Options: []oc.Option{
|
||||
{ID: OptionGoOn, Label: "Go on", Does: module + " counts healthy on " + node + " by your word and the delivery goes on", Level: occ.Acknowledge},
|
||||
{ID: OptionFailed, Label: "Count as failed", Does: "the failing check is the delivery's and is held as the canary rule has it", Level: occ.Acknowledge},
|
||||
},
|
||||
Who: occ.Operator,
|
||||
Expires: occ.FormatTime(now.Add(AskLasts)),
|
||||
OnExpiry: "the delivery keeps waiting",
|
||||
About: about,
|
||||
})
|
||||
}
|
||||
|
||||
// WaitsOutsideFinding is the finding `delivery-waits-outside` of a wait on an outside cause (design 47 G §706): scope delivery,
|
||||
// id `<node>.<module>`, severity warning, naming the module, the machine and the cause by its condition's key, and needing
|
||||
// nothing, so it is sent to no one: the operator's one message is the ask at the bound.
|
||||
func WaitsOutsideFinding(node, module, causeKey, on string, at time.Time) Observed {
|
||||
return Observed{Scope: "delivery", ID: node + "." + module, Kind: "delivery-waits-outside", Machine: node,
|
||||
Severity: "warning",
|
||||
Summary: fmt.Sprintf("%s on %s waits on %s (the condition %s)", module, node, on, causeKey),
|
||||
Headline: fmt.Sprintf("%s on %s waits on an outside cause", module, node),
|
||||
Explanation: fmt.Sprintf("The health check of %s fails on %s while %s is open, which is outside the delivery. The delivery "+
|
||||
"waits, judges again when it clears and goes on by itself.", module, node, causeKey),
|
||||
Needs: "Nothing for you to do.", Source: "assigner.delivery-waits-outside"}
|
||||
}
|
||||
|
||||
// HeldCondition is the condition the assigner raises when it holds a fingerprint of a module (ADR 0429 §6, ADR 0539 §19), or
|
||||
// clears (raised false) when the hold is released. It names the module, the node whose health check failed and the check's
|
||||
// words, under the assigner's own name (ADR 0437 clause 9), keyed by the held fingerprint. needsYou is true for a module holding
|
||||
|
||||
@@ -29,6 +29,7 @@ var writtenAgainOnEnd = map[string]string{
|
||||
asks.AskKindHold: "",
|
||||
asks.AskKindRestore: "",
|
||||
asks.AskKindBackup: "",
|
||||
asks.AskKindOutside: "", // raised again at once while the wait stands (design 47 G §706); the wait's end takes the key away
|
||||
asks.AskKindRecord: "",
|
||||
asks.AskKindAssignments: "a proposal's authorising ask that lapsed ends the proposal (ADR 0567 §5); the verb again asks it again (§11)",
|
||||
}
|
||||
|
||||
@@ -95,6 +95,9 @@ type Module struct {
|
||||
// source of a want's Self: every want the rules write or keep is derived from it (selfDerived), so a want written Self while
|
||||
// the node was placed does not outlive the assignments (issue 1126). Nil: no node does.
|
||||
Self func(node string) bool
|
||||
// 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
|
||||
}
|
||||
|
||||
func (m Module) selfOn(node string) bool { return m.Self != nil && m.Self(node) }
|
||||
@@ -126,6 +129,12 @@ func healthy(h nodeengine.HaveByNodeModuleValue, fp string) bool {
|
||||
return h.Fp == fp && h.Phase == nodeengineclient.PhaseRunning && h.Health == nodeengineclient.HealthOK
|
||||
}
|
||||
|
||||
// Healthy says whether a have runs fp healthy.
|
||||
func Healthy(h nodeengine.HaveByNodeModuleValue, fp string) bool { return healthy(h, fp) }
|
||||
|
||||
// Failed says whether a have's health check failed once the node is done moving.
|
||||
func Failed(h nodeengine.HaveByNodeModuleValue) bool { return failed(h) }
|
||||
|
||||
// Canary is the module's canary: the first of its nodes, by name, that is online (ADR 0435 §3; one ring, ordered by
|
||||
// name, ADR 0528 clause 3). Empty when none is online.
|
||||
func Canary(nodes []string, online map[string]bool) string {
|
||||
@@ -155,6 +164,8 @@ func holds(m Module, node string, n Nodes, ordering bool) (bool, int, string) {
|
||||
return false, mwc.RuleBuildDone, fmt.Sprintf("the build of %s at %s is not done", m.Name, short(m.FP))
|
||||
case m.Held:
|
||||
return false, mwc.RuleNotHeld, fmt.Sprintf("%s is held at %s", m.Name, short(m.FP))
|
||||
case m.MachineHeld != nil && m.MachineHeld(node):
|
||||
return false, mwc.RuleNotHeld, fmt.Sprintf("the condition machine.%s.network is open: no want moves to %s until its own network is healthy", node, node)
|
||||
}
|
||||
if ordering {
|
||||
canary := Canary(m.Nodes, n.Online)
|
||||
|
||||
@@ -22,6 +22,9 @@ const (
|
||||
Differences = plc.Module + "_" + wants.StateDifferences
|
||||
Asks = plc.Module + "_asks"
|
||||
Puts = plc.Module + "_puts"
|
||||
// Waits holds the waits on an outside cause, one key `<node>.<module>.<fingerprint hex>` (design 47 G §706).
|
||||
WaitsState = "waits"
|
||||
Waits = plc.Module + "_" + WaitsState
|
||||
// PoliciesState is the state of the policies a person set for a module's upgrade; Policies its bucket.
|
||||
PoliciesState = "policies"
|
||||
Policies = plc.Module + "_" + PoliciesState
|
||||
|
||||
@@ -0,0 +1,313 @@
|
||||
// Package outside is the assigner's wait on an outside cause (design 47 G §706, ADR 0677): a moved module's failing health
|
||||
// check on a machine whose open conditions name a cause outside the delivery is held as a wait, judged again on every change
|
||||
// that can end it, and ended by itself or by the operator. The package holds the rules and the kept record; the assigner's
|
||||
// process reads the conditions, says the events and writes the asks.
|
||||
package outside
|
||||
|
||||
import (
|
||||
"encoding/json"
|
||||
"fmt"
|
||||
"slices"
|
||||
"strings"
|
||||
"time"
|
||||
)
|
||||
|
||||
// The ways a wait ends, as `wait-ended` says them (§706 *How it ends*).
|
||||
const (
|
||||
HowHealthy = "healthy"
|
||||
HowCleared = "cleared"
|
||||
HowOwnFailure = "own-failure"
|
||||
HowGoOn = "go-on"
|
||||
HowFailed = "failed"
|
||||
HowReplaced = "replaced"
|
||||
// HowRejudge is a release by hand that asks to judge the wait again now; it ends nothing.
|
||||
HowRejudge = "rejudge"
|
||||
)
|
||||
|
||||
// OnUnread is what a wait stands on while the keeper does not answer.
|
||||
const OnUnread = "conditions-unread"
|
||||
|
||||
// The defaults of the settings (§633): `outside-wait-minutes` and `outside-settle-seconds`.
|
||||
const (
|
||||
DefaultBound = 30 * time.Minute
|
||||
DefaultSettle = 90 * time.Second
|
||||
)
|
||||
|
||||
// Cond is an open condition as the keeper's verb `conditions` answers it: its key and the subject's machine and `also`.
|
||||
type Cond struct {
|
||||
Key string
|
||||
Machine string
|
||||
Also []string
|
||||
}
|
||||
|
||||
// ParseConditions reads the keeper's answer of `conditions`.
|
||||
func ParseConditions(answer []byte) ([]Cond, error) {
|
||||
var a struct {
|
||||
Conditions []struct {
|
||||
Key string `json:"key"`
|
||||
Subject struct {
|
||||
Machine string `json:"machine"`
|
||||
Also []string `json:"also"`
|
||||
} `json:"subject"`
|
||||
} `json:"conditions"`
|
||||
}
|
||||
if err := json.Unmarshal(answer, &a); err != nil {
|
||||
return nil, fmt.Errorf("outside: the keeper's conditions do not read: %w", err)
|
||||
}
|
||||
out := make([]Cond, 0, len(a.Conditions))
|
||||
for _, c := range a.Conditions {
|
||||
out = append(out, Cond{Key: c.Key, Machine: c.Subject.Machine, Also: c.Subject.Also})
|
||||
}
|
||||
return out, nil
|
||||
}
|
||||
|
||||
// Cause is an outside cause open for a machine: the condition's key and what a wait says it stands `on`.
|
||||
type Cause struct{ Key, On string }
|
||||
|
||||
// Causes are the outside causes open for machine m among conds, sorted by key (§706 *What counts as an outside cause*):
|
||||
// `machine.M.<owner>.rewritten` or `machine.M.rewritten` unless the delivery moved <owner> to M (moved), and
|
||||
// `machine.<x>.unreachable`, `.network` or `.silent` of another machine whose `also` lists M. No other condition is one.
|
||||
func Causes(conds []Cond, m string, moved func(owner string) bool) []Cause {
|
||||
var out []Cause
|
||||
for _, c := range conds {
|
||||
p := strings.Split(c.Key, ".")
|
||||
if len(p) < 3 || p[0] != "machine" {
|
||||
continue
|
||||
}
|
||||
switch {
|
||||
case len(p) == 3 && p[1] == m && p[2] == "rewritten":
|
||||
out = append(out, Cause{c.Key, "rewritten"})
|
||||
case len(p) == 4 && p[1] == m && p[3] == "rewritten" && (moved == nil || !moved(p[2])):
|
||||
out = append(out, Cause{c.Key, "rewritten:" + p[2]})
|
||||
case len(p) == 3 && p[1] != m && slices.Contains([]string{"unreachable", "network", "silent"}, p[2]) && slices.Contains(c.Also, m):
|
||||
out = append(out, Cause{c.Key, p[2] + ":" + p[1]})
|
||||
}
|
||||
}
|
||||
slices.SortFunc(out, func(a, b Cause) int { return strings.Compare(a.Key, b.Key) })
|
||||
return out
|
||||
}
|
||||
|
||||
// OwnNetworkKey is the key of a machine's own open network condition, which holds the machine as a whole (§706 *M's own
|
||||
// network*).
|
||||
func OwnNetworkKey(m string) string { return "machine." + m + ".network" }
|
||||
|
||||
// OwnNetwork are the machines whose own network condition is open among conds.
|
||||
func OwnNetwork(conds []Cond) map[string]bool {
|
||||
out := map[string]bool{}
|
||||
for _, c := range conds {
|
||||
if p := strings.Split(c.Key, "."); len(p) == 3 && p[0] == "machine" && p[2] == "network" && c.Key == OwnNetworkKey(p[1]) {
|
||||
out[p[1]] = true
|
||||
}
|
||||
}
|
||||
return out
|
||||
}
|
||||
|
||||
// Wait is one outside wait as the assigner keeps it: the move of Module at FP on Node, or its settle (Settling). Decided is
|
||||
// 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"`
|
||||
Node string `json:"node"`
|
||||
Module string `json:"module"`
|
||||
FP string `json:"fp"`
|
||||
On string `json:"on,omitempty"`
|
||||
Cause string `json:"cause,omitempty"`
|
||||
Since time.Time `json:"since"`
|
||||
ClockFrom time.Time `json:"clock-from"`
|
||||
Bound string `json:"bound,omitempty"`
|
||||
// Settling: no outside cause was open at the failing check; the judgement is held until SettleEnds.
|
||||
Settling bool `json:"settling,omitempty"`
|
||||
SettleEnds time.Time `json:"settle-ends,omitempty"`
|
||||
// Ask is the id of the asking made at the bound, and State where the wait stands in words (settling, waiting, asked).
|
||||
Ask string `json:"ask,omitempty"`
|
||||
State string `json:"state,omitempty"`
|
||||
// Release is a release by hand not taken up yet: go-on, failed or rejudge, by By and why Why.
|
||||
Release string `json:"release,omitempty"`
|
||||
By string `json:"by,omitempty"`
|
||||
Why string `json:"why,omitempty"`
|
||||
// Decided is `go-on` or `failed` once the wait ended so.
|
||||
Decided string `json:"decided,omitempty"`
|
||||
}
|
||||
|
||||
// Key is the key of a wait in the state: `<node>.<module>.<fingerprint hex>`.
|
||||
func Key(node, module, hex string) string { return node + "." + module + "." + hex }
|
||||
|
||||
// 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.Node == "" || w.Module == "" {
|
||||
return w, fmt.Errorf("outside: %q is not a kept outside wait", raw)
|
||||
}
|
||||
return w, nil
|
||||
}
|
||||
|
||||
// Standing says a wait is said (`wait-held`) and not only settling, nor decided.
|
||||
func (w Wait) Standing() bool { return !w.Settling && w.Decided == "" }
|
||||
|
||||
// Facts are what a judging reads: the module's check on the node at the wait's fingerprint (Failing, Healthy), the keeper's
|
||||
// conditions as read now (Read false: the kept call is unanswered), whether the module holds the seat mesh-conditions itself,
|
||||
// and the settings.
|
||||
type Facts struct {
|
||||
Now time.Time
|
||||
Failing bool
|
||||
Healthy bool
|
||||
Read bool
|
||||
Conds []Cond
|
||||
Node string
|
||||
// Moved says the delivery moved the module <owner> to the node.
|
||||
Moved func(owner string) bool
|
||||
// CauseCleared: the condition the wait stands on was heard cleared (the keeper's event), so a healthy module ends `cleared`.
|
||||
CauseCleared bool
|
||||
// KeeperSelf: the module holds the seat mesh-conditions, which cannot answer for its own failing check.
|
||||
KeeperSelf bool
|
||||
Bound time.Duration
|
||||
Settle time.Duration
|
||||
}
|
||||
|
||||
// Decision is what a judging comes to.
|
||||
type Decision struct {
|
||||
// Wait is the record to keep: nil when none stands (none is kept; a decided one is kept with Decided).
|
||||
Wait *Wait
|
||||
// Held says wait-held is to be said (a new wait, or a new `on`); End is the `how` of a wait-ended to say, "" for none.
|
||||
Held bool
|
||||
End string
|
||||
// Own: the failure is the delivery's (the rules hold it as the canary rule has it). Counted: the module counts healthy
|
||||
// there by the operator's word. Waiting: the move waits (the rules see no failure, and the canary holds the rest).
|
||||
Own, Counted, Waiting bool
|
||||
// Ask: the bound has run from clock-from, so the operator is asked.
|
||||
Ask bool
|
||||
// Due is when the judging is due again by itself (the settle's end); zero for none.
|
||||
Due time.Time
|
||||
}
|
||||
|
||||
// Judge is one judging of the check at the node (§706). prev is the kept wait of that module at that fingerprint, nil for none.
|
||||
func Judge(prev *Wait, node, module, fp string, f Facts) Decision {
|
||||
if prev != nil && prev.Decided != "" {
|
||||
// Decided by the operator: it holds for this fingerprint until the module runs healthy.
|
||||
if f.Healthy {
|
||||
return Decision{}
|
||||
}
|
||||
d := Decision{Wait: prev, Own: prev.Decided == HowFailed, Counted: prev.Decided == HowGoOn}
|
||||
return d
|
||||
}
|
||||
if prev != nil && prev.Release != "" && prev.Release != HowRejudge {
|
||||
how := prev.Release
|
||||
w := *prev
|
||||
w.Release, w.Decided, w.Settling, w.Ask = "", how, false, ""
|
||||
d := Decision{Wait: &w, Own: how == HowFailed, Counted: how == HowGoOn}
|
||||
if prev.Standing() {
|
||||
d.End = how
|
||||
}
|
||||
return d
|
||||
}
|
||||
if f.Healthy {
|
||||
if prev == nil {
|
||||
return Decision{}
|
||||
}
|
||||
d := Decision{}
|
||||
if prev.Standing() {
|
||||
d.End = HowHealthy
|
||||
if prev.Cause != "" && (f.CauseCleared || (f.Read && !openKey(f.Conds, prev.Cause))) {
|
||||
d.End = HowCleared
|
||||
}
|
||||
}
|
||||
return d
|
||||
}
|
||||
if !f.Failing {
|
||||
// still moving: nothing is judged, what stands stands
|
||||
if prev == nil {
|
||||
return Decision{}
|
||||
}
|
||||
return Decision{Wait: prev, Waiting: prev.Standing()}
|
||||
}
|
||||
if f.KeeperSelf {
|
||||
d := Decision{Own: true}
|
||||
if prev != nil && prev.Standing() {
|
||||
d.End = HowOwnFailure
|
||||
}
|
||||
return d
|
||||
}
|
||||
next := Wait{Node: node, Module: module, FP: fp, Bound: f.Bound.String(), Since: f.Now, ClockFrom: f.Now}
|
||||
standing := prev != nil && prev.Standing()
|
||||
if standing {
|
||||
next = *prev
|
||||
next.Release = ""
|
||||
}
|
||||
if !f.Read {
|
||||
// The keeper does not answer. A judging that has only just asked holds, as the settle does; once the settle has run
|
||||
// from the failing check with no answer, the move waits on conditions-unread (§706 *When the keeper does not answer*).
|
||||
if standing {
|
||||
return stand(prev, *prev, f) // a wait that stands keeps its cause until the keeper answers
|
||||
}
|
||||
{
|
||||
if prev != nil && prev.Settling {
|
||||
next = *prev
|
||||
} else {
|
||||
next.Settling, next.SettleEnds, next.State = true, f.Now.Add(f.Settle), "settling"
|
||||
}
|
||||
if f.Now.Before(next.SettleEnds) {
|
||||
return Decision{Wait: &next, Waiting: true, Due: next.SettleEnds}
|
||||
}
|
||||
next.Settling, next.SettleEnds = false, time.Time{}
|
||||
}
|
||||
next.On, next.Cause, next.Settling, next.State = OnUnread, "", false, "waiting"
|
||||
return stand(prev, next, f)
|
||||
}
|
||||
causes := Causes(f.Conds, node, f.Moved)
|
||||
if len(causes) > 0 {
|
||||
c := causes[0]
|
||||
if standing {
|
||||
for _, k := range causes {
|
||||
if k.Key == prev.Cause {
|
||||
c = k
|
||||
}
|
||||
}
|
||||
}
|
||||
next.On, next.Cause, next.Settling, next.SettleEnds, next.State = c.On, c.Key, false, time.Time{}, "waiting"
|
||||
return stand(prev, next, f)
|
||||
}
|
||||
if standing {
|
||||
// judged again, failing, no outside cause open: the delivery's failure (after the settle the check has long had)
|
||||
w := *prev
|
||||
w.Decided, w.Release, w.Ask = HowOwnFailure, "", ""
|
||||
return Decision{Wait: &w, End: HowOwnFailure, Own: true}
|
||||
}
|
||||
// no outside cause open: settle from the failing check, judged again when it ends
|
||||
if prev != nil && prev.Settling {
|
||||
if !f.Now.Before(prev.SettleEnds) {
|
||||
w := *prev
|
||||
w.Settling, w.Decided = false, HowOwnFailure
|
||||
return Decision{Wait: &w, Own: true}
|
||||
}
|
||||
return Decision{Wait: prev, Waiting: true, Due: prev.SettleEnds}
|
||||
}
|
||||
next.Settling, next.SettleEnds, next.State, next.On, next.Cause = true, f.Now.Add(f.Settle), "settling", "", ""
|
||||
return Decision{Wait: &next, Waiting: true, Due: next.SettleEnds}
|
||||
}
|
||||
|
||||
// stand is the decision of a wait that stands on next.On: said when it is new or stands on another cause, its clock kept.
|
||||
func stand(prev *Wait, next Wait, f Facts) Decision {
|
||||
d := Decision{Wait: &next, Waiting: true}
|
||||
if prev == nil || !prev.Standing() || prev.On != next.On {
|
||||
d.Held = true
|
||||
}
|
||||
if prev != nil && prev.Standing() {
|
||||
next.Ask = prev.Ask
|
||||
}
|
||||
if f.Bound > 0 && !f.Now.Before(next.ClockFrom.Add(f.Bound)) {
|
||||
d.Ask = true
|
||||
next.State = "asked"
|
||||
}
|
||||
d.Wait = &next
|
||||
return d
|
||||
}
|
||||
|
||||
func openKey(conds []Cond, key string) bool {
|
||||
return slices.ContainsFunc(conds, func(c Cond) bool { return c.Key == key })
|
||||
}
|
||||
@@ -0,0 +1,169 @@
|
||||
package outside
|
||||
|
||||
import (
|
||||
"slices"
|
||||
"testing"
|
||||
"time"
|
||||
)
|
||||
|
||||
var t0 = time.Date(2026, 11, 11, 10, 0, 0, 0, time.UTC)
|
||||
|
||||
func TestCauses(t *testing.T) {
|
||||
conds := []Cond{
|
||||
{Key: "machine.ace.rewritten"},
|
||||
{Key: "machine.ace.systemd-resolved.rewritten"},
|
||||
{Key: "machine.ace.nats.rewritten"},
|
||||
{Key: "machine.hub.unreachable", Also: []string{"ace", "g14"}},
|
||||
{Key: "machine.lap.unreachable", Also: []string{"g14"}},
|
||||
{Key: "machine.hub2.network", Also: []string{"ace"}},
|
||||
{Key: "machine.hub3.silent", Also: []string{"ace"}},
|
||||
{Key: "machine.ace.network", Also: []string{"ace"}},
|
||||
{Key: "module.nats.ace.unhealthy", Machine: "ace"},
|
||||
}
|
||||
moved := func(o string) bool { return o == "nats" }
|
||||
var got []string
|
||||
for _, c := range Causes(conds, "ace", moved) {
|
||||
got = append(got, c.On)
|
||||
}
|
||||
want := []string{"unreachable:hub", "network:hub2", "silent:hub3", "rewritten", "rewritten:systemd-resolved"}
|
||||
slices.Sort(got)
|
||||
slices.Sort(want)
|
||||
if !slices.Equal(got, want) {
|
||||
t.Errorf("causes = %v, want %v (a rewritten naming a moved module, an unreachable not listing ace and ace's own network are none)", got, want)
|
||||
}
|
||||
if own := OwnNetwork(conds); !own["ace"] || !own["hub2"] || own["hub"] {
|
||||
t.Errorf("own network = %v", own)
|
||||
}
|
||||
}
|
||||
|
||||
func facts(now time.Time, conds ...Cond) Facts {
|
||||
return Facts{Now: now, Failing: true, Read: true, Conds: conds, Node: "ace", Bound: 30 * time.Minute, Settle: 90 * time.Second}
|
||||
}
|
||||
|
||||
func TestJudge(t *testing.T) {
|
||||
hub := Cond{Key: "machine.hub.unreachable", Also: []string{"ace"}}
|
||||
hubNet := Cond{Key: "machine.hub.network", Also: []string{"ace"}}
|
||||
step := func(prev *Wait, f Facts) (Decision, *Wait) {
|
||||
d := Judge(prev, "ace", "nats", "src2:aa", f)
|
||||
return d, d.Wait
|
||||
}
|
||||
tests := []struct {
|
||||
name string
|
||||
run func(t *testing.T)
|
||||
}{
|
||||
{"a failing check with a cause open is a wait said at once, ended cleared on its clearing", func(t *testing.T) {
|
||||
d, w := step(nil, facts(t0, hub))
|
||||
if !d.Held || !d.Waiting || w.On != "unreachable:hub" || w.Bound != "30m0s" || !w.ClockFrom.Equal(t0) || d.Own {
|
||||
t.Fatalf("decision %+v wait %+v", d, w)
|
||||
}
|
||||
f := facts(t0.Add(time.Minute))
|
||||
f.Failing, f.Healthy = false, true
|
||||
if d, _ := step(w, f); d.End != HowCleared || d.Wait != nil {
|
||||
t.Errorf("healthy once cleared: %+v", d)
|
||||
}
|
||||
f.Conds = []Cond{hub}
|
||||
if d, _ := step(w, f); d.End != HowHealthy {
|
||||
t.Errorf("healthy with the cause open: %+v", d)
|
||||
}
|
||||
}},
|
||||
{"a cause that does not list the machine leaves the failure to settle, then the delivery's", func(t *testing.T) {
|
||||
d, w := step(nil, facts(t0, Cond{Key: "machine.hub.unreachable", Also: []string{"g14"}}))
|
||||
if d.Held || !w.Settling || !w.SettleEnds.Equal(t0.Add(90*time.Second)) || !d.Waiting || d.Own {
|
||||
t.Fatalf("settle: %+v %+v", d, w)
|
||||
}
|
||||
if d, _ := step(w, facts(t0.Add(89*time.Second))); !d.Waiting || d.Own {
|
||||
t.Errorf("within the settle: %+v", d)
|
||||
}
|
||||
if d, _ := step(w, facts(t0.Add(90*time.Second))); !d.Own || d.Waiting {
|
||||
t.Errorf("past the settle: %+v", d)
|
||||
}
|
||||
}},
|
||||
{"a condition raised within the settle turns it into a wait", func(t *testing.T) {
|
||||
_, w := step(nil, facts(t0))
|
||||
d, w2 := step(w, facts(t0.Add(30*time.Second), hub))
|
||||
if !d.Held || w2.Settling || w2.On != "unreachable:hub" {
|
||||
t.Fatalf("%+v %+v", d, w2)
|
||||
}
|
||||
}},
|
||||
{"a cause that turns into another keeps the wait and its clock", func(t *testing.T) {
|
||||
_, w := step(nil, facts(t0, hub))
|
||||
d, w2 := step(w, facts(t0.Add(5*time.Minute), hubNet))
|
||||
if !d.Held || w2.On != "network:hub" || !w2.ClockFrom.Equal(t0) || d.End != "" {
|
||||
t.Fatalf("%+v %+v", d, w2)
|
||||
}
|
||||
}},
|
||||
{"judged again failing with no cause open ends own-failure", func(t *testing.T) {
|
||||
_, w := step(nil, facts(t0, hub))
|
||||
d, w2 := step(w, facts(t0.Add(time.Minute)))
|
||||
if d.End != HowOwnFailure || !d.Own || w2.Decided != HowOwnFailure {
|
||||
t.Fatalf("%+v", d)
|
||||
}
|
||||
}},
|
||||
{"the keeper silent holds as the settle does, then waits on conditions-unread, and is judged when it answers", func(t *testing.T) {
|
||||
f := facts(t0)
|
||||
f.Read = false
|
||||
d, w := step(nil, f)
|
||||
if d.Held || !w.Settling || !d.Waiting {
|
||||
t.Fatalf("%+v %+v", d, w)
|
||||
}
|
||||
f.Now = t0.Add(90 * time.Second)
|
||||
d, w = step(w, f)
|
||||
if !d.Held || w.On != OnUnread || w.Settling || !w.ClockFrom.Equal(t0) {
|
||||
t.Fatalf("%+v %+v", d, w)
|
||||
}
|
||||
if d, w2 := step(w, facts(t0.Add(100*time.Second), hub)); !d.Held || w2.On != "unreachable:hub" || !w2.ClockFrom.Equal(w.ClockFrom) {
|
||||
t.Errorf("answered: %+v %+v", d, w2)
|
||||
}
|
||||
}},
|
||||
{"the keeper's own module failing is its failure with no read", func(t *testing.T) {
|
||||
f := facts(t0, hub)
|
||||
f.KeeperSelf = true
|
||||
if d, w := step(nil, f); !d.Own || w != nil {
|
||||
t.Fatalf("%+v", d)
|
||||
}
|
||||
}},
|
||||
{"past the bound the operator is asked; Go on counts it healthy, Count as failed is the delivery's", func(t *testing.T) {
|
||||
_, w := step(nil, facts(t0, hub))
|
||||
d, w := step(w, facts(t0.Add(30*time.Minute), hub))
|
||||
if !d.Ask || d.Held || d.End != "" {
|
||||
t.Fatalf("%+v", d)
|
||||
}
|
||||
for how, want := range map[string]Decision{HowGoOn: {Counted: true}, HowFailed: {Own: true}} {
|
||||
w2 := *w
|
||||
w2.Release = how
|
||||
d, kept := step(&w2, facts(t0.Add(31*time.Minute), hub))
|
||||
if d.End != how || d.Counted != want.Counted || d.Own != want.Own || kept.Decided != how {
|
||||
t.Errorf("%s: %+v", how, d)
|
||||
}
|
||||
// decided: a check that goes on failing is not waited on again, until it runs healthy
|
||||
if d, _ := step(kept, facts(t0.Add(32*time.Minute), hub)); d.Held || d.Counted != want.Counted || d.Own != want.Own {
|
||||
t.Errorf("%s again: %+v", how, d)
|
||||
}
|
||||
}
|
||||
}},
|
||||
{"a release to judge again changes nothing by itself", func(t *testing.T) {
|
||||
_, w := step(nil, facts(t0, hub))
|
||||
w.Release = HowRejudge
|
||||
if d, w2 := step(w, facts(t0.Add(time.Minute))); d.End != HowOwnFailure || w2.Release != "" {
|
||||
t.Fatalf("%+v", d)
|
||||
}
|
||||
}},
|
||||
}
|
||||
for _, tt := range tests {
|
||||
t.Run(tt.name, tt.run)
|
||||
}
|
||||
}
|
||||
|
||||
func TestWaitRoundTrip(t *testing.T) {
|
||||
raw, err := Encode(Wait{Node: "ace", Module: "nats", FP: "src2:aa", On: "silent:g14", Since: t0, ClockFrom: t0})
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
w, err := Decode(raw)
|
||||
if err != nil || w.On != "silent:g14" || !w.ClockFrom.Equal(t0) {
|
||||
t.Fatalf("%+v %v", w, err)
|
||||
}
|
||||
if _, err := Decode([]byte(`{"v":2}`)); err == nil {
|
||||
t.Error("an unknown version is read")
|
||||
}
|
||||
}
|
||||
@@ -1,6 +1,7 @@
|
||||
package manifest
|
||||
|
||||
import (
|
||||
"encoding/json"
|
||||
"git.novox.be/novox/mesh-contracts/go/seats/meshmoduleassigner"
|
||||
"git.novox.be/novox/mesh-delivery/contracts/go/seats/meshwants"
|
||||
"git.novox.be/novox/mesh-delivery/modules/mesh-delivery-teller/tellerevents"
|
||||
@@ -65,7 +66,7 @@ func TestManifest(t *testing.T) {
|
||||
}
|
||||
if want := map[string]int{"asks": 5, "differences": 64, "puts": 20, "wants": 20,
|
||||
"assignments": 64, "placement": 64, "copies": 1, "proposals": 5, "candidates": 5, "pulses": 0, "findings": 5, "policies": 20, "seat-holders": 5, views.MarksState: 1, views.StatusState: 1,
|
||||
"engine-reads": 1, "written": 5}; !reflect.DeepEqual(states, want) {
|
||||
"engine-reads": 1, "written": 5, "waits": 5}; !reflect.DeepEqual(states, want) {
|
||||
t.Errorf("state = %v, want %v", states, want)
|
||||
}
|
||||
if _, ok := m.OwnSecrets["broker"]; !ok || len(m.OwnSecrets) != 1 {
|
||||
@@ -98,7 +99,7 @@ func TestManifest(t *testing.T) {
|
||||
// ADR 0621 §1: the first-view tools assignments, seat-holders and wants and design 25 § 12.2's six read tools, of the
|
||||
// program's bundle, which runs in a runtime of its own on the assigner's account; ADR 0618 §3: and assigner_engine_reads,
|
||||
// which lists and asks again what each node's engine reads.
|
||||
if !slices.Equal(m.Tools, []string{"assigner_put", "assigner_release", "assigner_puts", "assignments", "seat-holders", "wants",
|
||||
if !slices.Equal(m.Tools, []string{"assigner_put", "assigner_release", "assigner_puts", "assigner_waits", "assignments", "seat-holders", "wants",
|
||||
"assigner_read_assignments", "assigner_read_seat_holders", "assigner_read_wants", "assigner_read_findings", "assigner_read_asks", "assigner_read_differences", "assigner_read_copies",
|
||||
"assigner_first_views", "assigner_first_view_again", "assigner_engine_reads", "assigner_seat_list", "assigner_seat_list_again", "assigner_ask_again", "assigner_written"}) {
|
||||
t.Errorf("the assigner serves tools %v, want its three put tools, three first views, six read tools, assigner_engine_reads, the seat list's two (ADR 0636 §7) and assigner_ask_again (ADR 0668 §5)", m.Tools)
|
||||
@@ -128,17 +129,32 @@ 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-catalog.catalog-revision", "mesh-catalog.registration-changed", "mesh-catalog.seat-definition-changed", "mesh-catalog.seat-definition-removed", buckets.TellerModule + "." + tellerevents.AskEnded, "mesh-pipeline-runner.pipeline-ended", "mesh-pipeline-runner.group-ended", "node-engine.manifest-reads-changed", "node-engine.running"}) {
|
||||
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.seat-definition-changed", "mesh-catalog.seat-definition-removed", buckets.TellerModule + "." + tellerevents.AskEnded, "mesh-pipeline-runner.pipeline-ended", "mesh-pipeline-runner.group-ended", "node-engine.manifest-reads-changed", "node-engine.running"}) {
|
||||
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",
|
||||
"seat:mesh-catalog.registrations", "seat:mesh-catalog.revision", "seat:mesh-catalog.seat-definitions", // ADR 0636 §4: the seat list
|
||||
"seat:mesh-conditions.conditions", // design 47 G §706: the conditions read at each judging of an outside wait
|
||||
"seat:mesh-hand-acts.hand-act", "seat:mesh-hand-acts.hand-acts", "seat:mesh-hand-acts.warranted", // ADR 0565 §9, §11: the seat alone
|
||||
"seat:node-backup.backed-up", "seat:node-backup.now",
|
||||
"seat:node-engine.manifest-reads", // ADR 0618 §3
|
||||
"seat:operator-channel.asks"}; !slices.Equal(slices.Sorted(slices.Values(m.Invokes)), want) { // ADR 0668 §4: the ended asks at bind
|
||||
t.Errorf("invokes = %v, want %v", m.Invokes, want)
|
||||
}
|
||||
// design 47 G §633, §706: the bounds of the wait on an outside cause are `config` keys filling the process's environment.
|
||||
var raw struct {
|
||||
Config map[string]struct {
|
||||
Default any `json:"default"`
|
||||
} `json:"config"`
|
||||
}
|
||||
file, _ := os.ReadFile("module.json")
|
||||
if err := json.Unmarshal(file, &raw); err != nil || raw.Config["outside-wait-minutes"].Default != float64(30) || raw.Config["outside-settle-seconds"].Default != float64(90) || len(raw.Config) != 2 {
|
||||
t.Errorf("config = %+v (%v), want outside-wait-minutes of 30 and outside-settle-seconds of 90", raw.Config, err)
|
||||
}
|
||||
if !strings.Contains(string(file), `"MESH_ASSIGNER_OUTSIDE_WAIT_MINUTES": "${config:outside-wait-minutes}"`) ||
|
||||
!strings.Contains(string(file), `"MESH_ASSIGNER_OUTSIDE_SETTLE_SECONDS": "${config:outside-settle-seconds}"`) {
|
||||
t.Errorf("the process's env does not carry the two settings from config")
|
||||
}
|
||||
if want := []string{"assignments-copied", "put", "views.asks", "views.copies", "views.differences", "views.findings", "views.assignments", "views.seat-holders", "views.wants"}; !slices.Equal(m.Emits, want) {
|
||||
t.Errorf("emits = %v, want the copy, put and its six views", m.Emits)
|
||||
}
|
||||
|
||||
@@ -43,6 +43,9 @@
|
||||
],
|
||||
"consumes": [
|
||||
"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.seat-definition-changed",
|
||||
@@ -59,6 +62,7 @@
|
||||
"seat:mesh-catalog.registrations",
|
||||
"seat:mesh-catalog.revision",
|
||||
"seat:mesh-catalog.seat-definitions",
|
||||
"seat:mesh-conditions.conditions",
|
||||
"seat:mesh-hand-acts.hand-act",
|
||||
"seat:mesh-hand-acts.hand-acts",
|
||||
"seat:mesh-hand-acts.warranted",
|
||||
@@ -90,6 +94,7 @@
|
||||
"assigner_put",
|
||||
"assigner_release",
|
||||
"assigner_puts",
|
||||
"assigner_waits",
|
||||
"assignments",
|
||||
"seat-holders",
|
||||
"wants",
|
||||
@@ -125,8 +130,27 @@
|
||||
{ "name": "view-marks", "history": 1 },
|
||||
{ "name": "first-views", "history": 1 },
|
||||
{ "name": "engine-reads", "history": 1 },
|
||||
{ "name": "written", "history": 5 }
|
||||
{ "name": "written", "history": 5 },
|
||||
{ "name": "waits", "history": 5 }
|
||||
],
|
||||
"config": {
|
||||
"outside-wait-minutes": {
|
||||
"type": "number",
|
||||
"description": "minutes a moved module's failing check waits on an outside cause before the operator is asked to go on or count it as failed (design 47 G section 706); a bound a whole test sets in seconds.",
|
||||
"minimum": 1,
|
||||
"maximum": 10080,
|
||||
"trusted": false,
|
||||
"default": 30
|
||||
},
|
||||
"outside-settle-seconds": {
|
||||
"type": "number",
|
||||
"description": "seconds the assigner holds its judgement of a failing check when no outside cause is open yet, since the node-engine states the failing check and the network in one report while the condition rises only after the watchman's two looks and the keeper's take-in (design 47 G section 706).",
|
||||
"minimum": 1,
|
||||
"maximum": 3600,
|
||||
"trusted": false,
|
||||
"default": 90
|
||||
}
|
||||
},
|
||||
"objects": [
|
||||
{ "name": "assignment-revisions", "max-mib": 512, "form": 1 }
|
||||
],
|
||||
@@ -159,7 +183,9 @@
|
||||
"./mesh-delivery-assigner"
|
||||
],
|
||||
"env": {
|
||||
"MESH_BROKER_FILE": "${dir:state}/broker"
|
||||
"MESH_BROKER_FILE": "${dir:state}/broker",
|
||||
"MESH_ASSIGNER_OUTSIDE_WAIT_MINUTES": "${config:outside-wait-minutes}",
|
||||
"MESH_ASSIGNER_OUTSIDE_SETTLE_SECONDS": "${config:outside-settle-seconds}"
|
||||
},
|
||||
"user": "mesh-delivery-assigner",
|
||||
"health": {
|
||||
|
||||
Reference in New Issue
Block a user