Files
mesh-delivery/modules/mesh-delivery-view/internal/view/view.go
T
jschoubben da8384d10e VW5: the view takes each watched branch's source from the gate's verb source, per branch
The view read the gate's state source and its history, fixed to the branch main. Each watched branch's source is now
the seat mesh-delivery's verb source given that branch, asked again on its source-moved; the source row and the event
delivered are per branch, a repository's reach is from the source of its modules' registration's first branch, laid
over the reach the view's own state kept at its start, and view_tiers given no revision answers per branch. The
manifest drops its read of mesh-delivery-gate.source and names the source answer's fields it relies on.

Builds: design 47 G §507, §512, §532, §558, A-173, A-225

Claude-Session: https://claude.ai/code/session_01NcbgfGKaxw984XyAPfLm4h
2026-11-12 06:21:49 +01:00

1357 lines
50 KiB
Go

// Package view derives the operator's rows from the delivery buckets.
package view
import (
"fmt"
"maps"
"slices"
"sort"
"strings"
"sync"
"time"
"git.novox.be/novox/mesh-contracts/go/clients/meshmoduleassignerclient"
"git.novox.be/novox/mesh-contracts/go/clients/nodeengineclient"
"git.novox.be/novox/mesh-contracts/go/seats/nodeengine"
"git.novox.be/novox/mesh-contracts/health-and-repair/go/clients/meshwatchdogsclient"
occ "git.novox.be/novox/mesh-contracts/operator-and-conversation/go/clients/operatorchannelclient"
oc "git.novox.be/novox/mesh-contracts/operator-and-conversation/go/seats/operatorchannel"
"git.novox.be/novox/mesh-delivery/modules/mesh-delivery-assigner/asks"
"git.novox.be/novox/mesh-delivery/modules/mesh-delivery-builder/builds"
"git.novox.be/novox/mesh-delivery/modules/mesh-delivery-contracts/go/clients/meshcatalogclient"
"git.novox.be/novox/mesh-delivery/modules/mesh-delivery-contracts/go/clients/meshdeliveryclient"
"git.novox.be/novox/mesh-delivery/modules/mesh-delivery-contracts/go/clients/meshwantsclient"
"git.novox.be/novox/mesh-delivery/modules/mesh-delivery-contracts/go/seats/meshcatalog"
"git.novox.be/novox/mesh-delivery/modules/mesh-delivery-indexer/takein"
"git.novox.be/novox/mesh-delivery/modules/mesh-delivery-sdk/buildrecord"
"git.novox.be/novox/mesh-delivery/modules/mesh-delivery-sdk/trunk"
"git.novox.be/novox/mesh-delivery/modules/mesh-delivery-view/viewstate"
"git.novox.be/novox/mesh-sdk/go/keys"
"git.novox.be/novox/mesh-sdk/go/motion"
)
// Entry is one change of a bucket's key, as the watch delivers it.
type Entry struct {
Bucket string
Key string
Value []byte
At time.Time
Deleted bool
// Revision is the bucket revision of the change, when the watch gives one; for a branch's source (SourceBucket, keyed by the
// escaped branch) it is the source revision the gate's verb source answers.
Revision uint64
// Branch is, for a module's entry (ModulesBucket), the watched branch it is read from: the first branch of the module's
// registration, before the first registrations the first branch the gate's source answers that lists it (design 47 G A-173).
Branch string
// History says the change comes from a watch with history (ReachHistory): it feeds only the history the reach reads.
History bool
}
// Row is one module on one node.
type Row struct {
Node string `json:"node"`
Module string `json:"module"`
Status string `json:"status"`
Assigned string `json:"assigned,omitempty"`
Have string `json:"have,omitempty"`
Phase string `json:"phase,omitempty"`
Health string `json:"health,omitempty"`
Since time.Time `json:"since,omitzero"`
Detail string `json:"detail,omitempty"`
Waits string `json:"waits,omitempty"`
// RequiredBy is what pulled the module onto the node, for an assignment that is pulled in.
RequiredBy string `json:"required-by,omitempty"`
Error string `json:"error,omitempty"`
}
// NodeRow is a node's own facts as counts.
type NodeRow struct {
Node string `json:"node"`
Windows int `json:"windows"`
Busy int `json:"busy"`
Strays int `json:"strays"`
Since time.Time `json:"since,omitzero"`
Error string `json:"error,omitempty"`
}
// Snapshot is everything the view shows.
type Snapshot struct {
Rows []Row `json:"rows"`
Nodes []NodeRow `json:"nodes"`
// Offline is every node and part instance whose presence is offline (the watchman's word).
Offline []OfflineRow `json:"offline"`
// Stuck is every key that stopped moving and has not moved again.
Stuck []StuckRow `json:"stuck"`
// Unreadable is every watchman value the view could not read, named by its key: never dropped silently.
Unreadable []UnreadableRow `json:"unreadable"`
// Groups is every candidate group the indexer holds: the pull requests that share a branch, with the modules each touches.
Groups []GroupRow `json:"groups"`
// Fingerprints is every module the indexer fingerprinted at a trunk merge, beside its newest successful trunk build's.
Fingerprints []FingerprintRow `json:"fingerprints"`
// Asks is every ask the gate holds that has no answer yet: what waits for the operator's Approve.
Asks []AskRow `json:"asks"`
// Wants is every want of the assigner beside the have of its node and where it stands (novox/hq ADR 0539 §8, issue 1906 D7).
Wants []WantRow `json:"wants"`
// Sources is the state of each watched branch's source revision, derived from the wants and the haves (ADR 0429 §1;
// design 47 G A-173): recorded by no part, by branch name. Empty while the gate's verb source answers no branch's.
Sources []SourceRow `json:"sources,omitempty"`
// Backlog is, for each language and kind of build ask with one waiting or held, the asks waiting for a node, the asks a node
// holds and the oldest waiting ask's age (issue 1887 A8a).
Backlog []BacklogRow `json:"backlog"`
}
// The states of a want against the have of its node.
const (
WantApplied = "applied" // the node runs the want's fingerprint, running and healthy
WantPending = "pending" // the node has not yet run the want's fingerprint, or not healthy yet
WantFailed = "failed" // the node runs the want's fingerprint and its health check failed
WantHeldBack = "held back" // the assigner holds the node back from the newest fingerprint (Rule), or set it back
WantRemoved = "removed" // the node's want is removed: the module is no longer assigned to it
)
// WantRow is a want of the assigner beside the have of its node: the fingerprint wanted, the one the node runs with its phase and
// health, the state of the two, and for a want held back the newest fingerprint, the rule and the assigner's words.
type WantRow struct {
Node string `json:"node"`
Module string `json:"module"`
Want string `json:"want,omitempty"`
Have string `json:"have,omitempty"`
Phase string `json:"phase,omitempty"`
Health string `json:"health,omitempty"`
State string `json:"state"`
Newest string `json:"newest,omitempty"`
Rule int `json:"rule,omitempty"`
Words string `json:"words,omitempty"`
// Judgement is the node agent's judgement of the move that waits for a person's yes (its have in waiting-yes, ADR 0547
// §3), and Judged its words; empty otherwise. The controller's composed copy is no longer read (ADR 0571 §6).
Judgement string `json:"judgement,omitempty"`
Judged string `json:"judged,omitempty"`
}
// The states of a source revision (ADR 0429 §1).
const (
SourceDelivered = "delivered"
SourceNotDelivered = "not delivered"
)
// SourceRow is one watched branch's source revision (as the gate's verb source answers it) and whether it is delivered: every
// want of a module whose branch it is equals its module's fingerprint in the catalog half and every have equals its want, running
// and healthy, and every node such a module is assigned to has a want. Waiting names each module on a node that is not there
// yet, and why.
type SourceRow struct {
Branch string `json:"branch"`
Revision uint64 `json:"revision"`
State string `json:"state"`
Waiting []string `json:"waiting,omitempty"`
// Wants is how many wants of modules at the indexer were looked at; a revision with none is delivered of nothing.
Wants int `json:"wants"`
// Away are the nodes the watchman says are offline, left out of the comparison (ADR 0603 §7): the revision counts for them
// once they return and run their wants.
Away []string `json:"away,omitempty"`
}
// AskRow is an open ask (novox/hq ADR 0437 §10): the gate's ask with no answer under its key in the teller's answers.
type AskRow struct {
Key string `json:"key"`
Group string `json:"group"`
Headline string `json:"headline"`
Words string `json:"words"`
Pulls []string `json:"pulls"`
Expires time.Time `json:"expires,omitzero"`
How string `json:"how"`
// Options is what the view offers to answer with, for an ask that needs no proof of who answered (the assigner's hold and
// restore, ADR 0528 clause 6); Chosen is the option the view wrote that the teller has not filled yet.
Options []string `json:"options,omitempty"`
Chosen string `json:"chosen,omitempty"`
}
// AnswerInTheView is what an open ask that needs no proof of who answered says (ADR 0528 clause 6): the operator answers it here.
const AnswerInTheView = "answer in the view"
// GroupRow is a candidate group (novox/hq ADR 0437 §5): the branch its pull requests share and the pull requests.
type GroupRow struct {
Group string `json:"group"`
Pulls []GroupPull `json:"pulls"`
}
// GroupPull is one pull request of a group, its head shortened to 12 digits, and the modules it touches.
type GroupPull struct {
Repository string `json:"repository"`
Number int `json:"number"`
Head string `json:"head"`
Modules []string `json:"modules"`
}
// The states of a module's fingerprint against its newest successful trunk build's.
const (
FPSame = "same"
FPDiffers = "differs"
FPNone = "none"
FPNoBuild = "no build"
// The states of a fingerprint of the new form against the builder's key of it (I5, issue 1851): built (the key is done),
// build failed (Reason is its error), building (queued, claimed or building; Reason says what it waits for) and no key.
FPBuilt = "built"
FPBuildFailed = "build failed"
FPBuilding = "building"
FPNoKey = "no key"
)
// FingerprintRow compares the indexer's fingerprint of a module at a trunk merge with the source fingerprint its newest
// successful trunk build recorded. State is same, differs (both are given), none (the indexer could not fingerprint it:
// Reason says why) or no build (the indexer's alone is given, no successful trunk build recorded one).
type FingerprintRow struct {
Module string `json:"module"`
Repository string `json:"repository,omitempty"`
Commit string `json:"commit,omitempty"`
State string `json:"state"`
Indexer string `json:"indexer,omitempty"`
Build string `json:"build,omitempty"`
Reason string `json:"reason,omitempty"`
}
// OfflineRow is a node ("node") or a part's instance ("part") that is offline, since the presence entry's time.
type OfflineRow struct {
Kind string `json:"kind"`
Node string `json:"node,omitempty"`
Part string `json:"part,omitempty"`
Instance string `json:"instance,omitempty"`
Since time.Time `json:"since,omitzero"`
}
// StuckRow is a key that stopped moving: its bucket and key, what it waited for and since when.
type StuckRow struct {
Bucket string `json:"bucket"`
Key string `json:"key"`
Waited string `json:"waited,omitempty"`
Since time.Time `json:"since,omitzero"`
}
// UnreadableRow is a value of the watchman's buckets that did not decode, with the reader's error.
type UnreadableRow struct {
Bucket string `json:"bucket"`
Key string `json:"key"`
Error string `json:"error"`
}
// Model holds the rows' inputs in memory and nothing else: a restart fills it again from the buckets' last values.
type Model struct {
mu sync.Mutex
placed *meshmoduleassignerclient.Placed // the seat's assignments (the assigner's state, ADR 0567, ADR 0549 §3)
builds map[string]buildEntry // by <module>.build.<id>
groups map[string]groupEntry // the indexer's candidates, by group
modules map[string]moduleEntry // the indexer's modules, by module
keys map[string]keyEntry // the builder's build keys, by <module>.<fingerprint>
wants map[string]wantEntry // the assigner's wants, by <node>.<module>
sources map[string]*sourceEntry // each watched branch's source, by escaped branch (design 47 G A-173)
modBranch map[string]string // the escaped branch each module's entry is read from, by module
deliveredEvent func(DeliveredEvent) error
motionEvent func(MotionEvent) error
told map[string]bool // the motion events said, by their ids
initial map[string]bool // the buckets whose watch handed over what stood at its start
said map[string]uint64 // the source revision whose delivery was said, by escaped branch
aasks map[string]assignerAskRecord // the assigner's asks (hold, restore), by key
chosen map[string]viewstate.ViewAnswer // the answers the view wrote, by ask key
writer Writer
backlog *Backlog
answers map[string]string // by key: the router id of the asking the teller's answer is for ("" when unreadable)
answerOf map[string]meshdeliveryclient.Answer // by key: the teller's answer whole, when readable (holds.go)
holds map[string]heldEntry // the assigner's holds, by held key (holds.go)
newest map[string]string // module -> id
haves map[string]haveEntry
nodes map[string]nodeEntry
// the watchman's buckets, by key
presence map[string]presenceEntry
stuck map[string]stuckEntry
// how far each repository reached (reach.go, ADR 0556)
modHist map[string][]meshcatalog.ModulesByModuleValue // the indexer's values of each module, oldest first
srcHist map[string]map[uint64]map[string]string // each branch's source heads heard since the start, by escaped branch and revision
kept map[string]viewstate.Reach // the reach the view's own state held at its start, by repository key (A-225)
taken map[string]takein.Taken // the indexer's taken, by repository key
reach map[string]viewstate.Reach // the reach derived last, by repository key
reachSaid map[string]uint64 // the reached head's source revision said last, by repository key
reachNotYet map[string]string // the merges said not yet on the machines with it, by repository key
reachPut func(ReachWrite) error
}
type presenceEntry struct {
value meshwatchdogsclient.NodePresence
err string
at time.Time
}
type stuckEntry struct {
value meshwatchdogsclient.Stuck
err string
}
// pair is a module the seat's assignments puts on a node.
type pair struct{ Node, Module string }
// placedPairs is every module the seat's assignments puts on a node, by node then module. The caller holds mu.
func (m *Model) placedPairs() []pair {
byNode := m.placed.ByNode()
var out []pair
for _, node := range slices.Sorted(maps.Keys(byNode)) {
for _, module := range byNode[node] {
out = append(out, pair{Node: node, Module: module})
}
}
return out
}
// build holds only the fields of the controller's Build record the view reads. A record with no state and no failure
// is a done build.
type build struct {
ID string `json:"id"`
Module string `json:"module"`
State string `json:"state"`
Commit string `json:"commit"`
Failed string `json:"failed"`
}
func (b build) done() bool { return b.State == "" && b.Failed == "" }
// buildEntry is a build as the view reads it: its own fields (state, failure) and the SDK's record (the source
// fingerprint, the ref, the time), decoded as the controller writes them.
type buildEntry struct {
build
rec buildrecord.BuildRecord
}
type groupEntry struct {
value meshcatalog.CandidatesByGroupValue
err string
}
type keyEntry struct {
value builds.BuilderBuild
err string
}
type wantEntry struct {
value meshwantsclient.Want
err string
at time.Time
}
type sourceEntry struct {
branch string // the branch's name
revision uint64
group string
heads map[string]string
err string
}
type assignerAskRecord struct {
ask oc.Ask
err string
}
type moduleEntry struct {
value meshcatalog.ModulesByModuleValue
err string
}
type haveEntry struct {
have nodeengine.HaveByNodeModuleValue
err string
at time.Time
}
type nodeEntry struct {
node nodeengine.HaveByNodeNodeValue
err string
at time.Time
}
// New is an empty model.
func New() *Model {
return &Model{placed: meshmoduleassignerclient.NewPlaced(), builds: map[string]buildEntry{}, groups: map[string]groupEntry{}, modules: map[string]moduleEntry{}, keys: map[string]keyEntry{}, wants: map[string]wantEntry{}, told: map[string]bool{}, initial: map[string]bool{}, aasks: map[string]assignerAskRecord{}, chosen: map[string]viewstate.ViewAnswer{}, answers: map[string]string{}, answerOf: map[string]meshdeliveryclient.Answer{}, holds: map[string]heldEntry{}, newest: map[string]string{},
haves: map[string]haveEntry{}, nodes: map[string]nodeEntry{},
presence: map[string]presenceEntry{}, stuck: map[string]stuckEntry{},
modHist: map[string][]meshcatalog.ModulesByModuleValue{}, srcHist: map[string]map[uint64]map[string]string{}, kept: map[string]viewstate.Reach{},
sources: map[string]*sourceEntry{}, modBranch: map[string]string{}, said: map[string]uint64{}, taken: map[string]takein.Taken{},
reach: map[string]viewstate.Reach{}, reachSaid: map[string]uint64{}, reachNotYet: map[string]string{}}
}
// Apply takes one change of a key. A value it cannot read is kept as an error on its row, never guessed. When the change makes
// the source revision delivered, the event of it is said once (SetDeliveredEvent), after the model's lock is released.
func (m *Model) Apply(e Entry) {
m.mu.Lock()
var events []DeliveredEvent
var motion []MotionEvent
if e.History {
m.applyHistory(e)
} else {
m.apply(e)
events = m.deliveredNow()
motion = m.motionNow(e)
}
writes, reached := m.reachNow()
say, put := m.deliveredEvent, m.reachPut
sayMotion := m.motionEvent
m.mu.Unlock()
for _, event := range events {
if say != nil && say(event) != nil {
m.unsay(event)
}
}
m.sendReach(put, say, writes, reached)
m.sayMotion(sayMotion, motion)
}
// sayMotion says the motion events after the model's lock is released; one not said is forgotten as told, so the next change of
// its want, have or stuck key says it again.
func (m *Model) sayMotion(say func(MotionEvent) error, events []MotionEvent) {
if say == nil {
return
}
for _, ev := range events {
if say(ev) != nil {
m.mu.Lock()
delete(m.told, ev.ID)
m.mu.Unlock()
}
}
}
// sendReach writes the reach values and says the reached heads, after the model's lock is released. A value not written or an
// event not said is forgotten as written or said, so the next change of any key the view watches writes or says it again
// (issue 1126: a write refused while the bus was not yet answering left the view's state reach at an older have for good, and
// mesh-delivery.show and knowledge-base.progress said a module not running that node_haves said running).
func (m *Model) sendReach(put func(ReachWrite) error, say func(DeliveredEvent) error, writes []ReachWrite, reached []DeliveredEvent) {
var unwritten []ReachWrite
var unsaid []DeliveredEvent
if put != nil {
for _, w := range writes {
if put(w) != nil {
unwritten = append(unwritten, w)
}
}
}
if say != nil {
for _, ev := range reached {
if say(ev) != nil {
unsaid = append(unsaid, ev)
}
}
}
if len(unwritten) == 0 && len(unsaid) == 0 {
return
}
m.mu.Lock()
defer m.mu.Unlock()
for _, w := range unwritten {
r, err := viewstate.DecodeReach(w.Value)
if prev, ok := m.reach[w.Key]; ok && err == nil && viewstate.SameReach(prev, r) {
delete(m.reach, w.Key) // a newer value derived since stands as it is: its own write says whether it landed
}
}
for _, ev := range unsaid {
key, err := trunk.RepositoryKey(ev.Repository)
if err == nil && m.reachSaid[key] == ev.Revision && m.reachNotYet[key] == strings.Join(ev.NotYet, ",") {
m.reachNotYet[key] = unsaidMark
}
}
}
// unsaidMark stands for a reached head's event not said: it equals no list of merges, so the next derivation says it again.
const unsaidMark = "\x00not said"
// unsay forgets that a source revision's delivered event was said, when it could not be said and nothing newer was said since.
func (m *Model) unsay(ev DeliveredEvent) {
m.mu.Lock()
defer m.mu.Unlock()
if b := keys.EscapeName(ev.Branch); m.said[b] == ev.Revision {
delete(m.said, b)
}
}
// DeliveredEvent is what the view says when the source revision is delivered (event `delivered` of the view, issue 1906 D9, ADR
// 0539 §17): the revision of the gate's source key, the group that moved it and every repository's merged head at it. A
// consumer takes every earlier revision as covered by it, and takes the same revision twice as once (after a restart of the view).
//
// The same event says a repository's reached head (ADR 0556 clause 3): Repository names it, Heads holds its head alone, Revision
// is the source revision that named that head and Group is empty; a consumer takes every earlier merge of that repository as on
// the machines.
type DeliveredEvent struct {
// Branch is the watched branch whose source revision it is (design 47 G A-173).
Branch string `json:"branch"`
Revision uint64 `json:"revision"`
Group string `json:"group"`
Heads map[string]string `json:"heads"`
Repository string `json:"repository,omitempty"`
// NotYet are, in the per-repository form, the merges at or before the reached head that are not yet on the machines: the
// changes of a module a node wants and runs nothing countable of. Said again when they change.
NotYet []string `json:"not_yet,omitempty"`
}
// SetDeliveredEvent sets where the event of a delivered source revision goes; nothing is said until it is set.
func (m *Model) SetDeliveredEvent(say func(DeliveredEvent) error) {
m.mu.Lock()
defer m.mu.Unlock()
m.deliveredEvent = say
}
// deliveredNow is the event of each watched branch's source revision that is delivered now and has not been said, in branch
// order. A revision that went bad after it was delivered and recovered is not said again.
func (m *Model) deliveredNow() []DeliveredEvent {
if m.deliveredEvent == nil {
return nil
}
var out []DeliveredEvent
for _, b := range slices.Sorted(maps.Keys(m.sources)) {
src := m.sources[b]
if src.err != "" || m.said[b] == src.revision {
continue
}
row := m.sourceRow(b)
if row == nil || row.State != SourceDelivered || row.Wants == 0 {
continue
}
m.said[b] = src.revision
out = append(out, DeliveredEvent{Branch: src.branch, Revision: src.revision, Group: src.group, Heads: src.heads})
}
return out
}
func (m *Model) apply(e Entry) {
switch e.Bucket {
case HaveBucket:
node, _, isNode, err := nodeengineclient.ParseHaveKey(e.Key)
if err != nil {
return
}
if isNode {
if e.Deleted {
delete(m.nodes, node)
return
}
n, err := nodeengineclient.DecodeHaveNode(e.Value)
m.nodes[node] = nodeEntry{node: n, err: errText(err), at: e.At}
return
}
key := e.Key // the bucket's own key: a module name with a dot is escaped in it (issue 1150)
if e.Deleted {
delete(m.haves, key)
return
}
h, err := nodeengineclient.DecodeHave(e.Value)
m.haves[key] = haveEntry{have: h, err: errText(err), at: e.At}
case PresenceBucket:
if e.Deleted {
delete(m.presence, e.Key)
return
}
p, err := meshwatchdogsclient.DecodeNodePresence(e.Value)
m.presence[e.Key] = presenceEntry{value: p, err: errText(err), at: e.At}
case StuckBucket:
if e.Deleted {
delete(m.stuck, e.Key)
return
}
st, err := meshwatchdogsclient.DecodeStuck(e.Value)
m.stuck[e.Key] = stuckEntry{value: st, err: errText(err)}
case AssignmentsBucket:
// A key that does not read places nothing on its node; the seat's verb assignments says why.
_, _, _ = m.placed.Put(e.Key, e.Value, e.Deleted, e.At)
case CandidatesBucket:
if e.Deleted {
delete(m.groups, e.Key)
return
}
c, err := meshcatalog.DecodeCandidatesByGroupValue(e.Value)
m.groups[e.Key] = groupEntry{value: c, err: errText(err)}
case ModulesBucket:
if e.Deleted {
delete(m.modules, e.Key)
delete(m.modBranch, e.Key)
return
}
mod, err := meshcatalogclient.DecodeModule(e.Value)
m.modules[e.Key] = moduleEntry{value: mod, err: errText(err)}
if e.Branch != "" {
m.modBranch[e.Key] = keys.EscapeName(e.Branch)
} else {
delete(m.modBranch, e.Key)
}
case AssignerAsksBucket:
if e.Deleted {
delete(m.aasks, e.Key)
return
}
a, err := decodeAssignerAsk(e.Key, e.Value)
m.aasks[e.Key] = assignerAskRecord{ask: a, err: errText(err)}
case viewstate.ViewAnswersBucket:
askKey, err := viewstate.ParseViewAnswerKey(e.Key)
if err != nil {
return
}
if e.Deleted {
delete(m.chosen, askKey)
return
}
if a, err := viewstate.DecodeViewAnswer(e.Value); err == nil {
m.chosen[askKey] = a
}
case AssignerWantsBucket:
if m.applyHeld(e) {
return
}
node, module, err := meshwantsclient.ParseWantKey(e.Key)
if err != nil {
return // the node's own key is not a module's want
}
key, kerr := nodeengineclient.HaveKey(node, module)
if kerr != nil {
return
}
if e.Deleted {
delete(m.wants, key)
return
}
w, err := meshwantsclient.DecodeWant(e.Value)
m.wants[key] = wantEntry{value: w, err: errText(err), at: e.At}
case TakenBucket:
if e.Deleted {
delete(m.taken, e.Key)
return
}
if t, err := takein.DecodeTaken(e.Value); err == nil {
m.taken[e.Key] = t
} else {
delete(m.taken, e.Key)
}
case SourceBucket: // one watched branch's source, keyed by the escaped branch (design 47 G A-173)
if e.Deleted {
delete(m.sources, e.Key)
delete(m.said, e.Key)
return
}
name, err := keys.UnescapeName(e.Key)
if err != nil {
return
}
src, err := meshdeliveryclient.DecodeSource(e.Value)
m.sources[e.Key] = &sourceEntry{branch: name, revision: e.Revision, group: src.Group, heads: src.Heads, err: errText(err)}
if err == nil && e.Revision != 0 { // the history the reach reads: each revision heard since the start (A-225)
if m.srcHist[e.Key] == nil {
m.srcHist[e.Key] = map[uint64]map[string]string{}
}
m.srcHist[e.Key][e.Revision] = src.Heads
}
case viewstate.ReachBucket: // the view's own state reach, as it stood at the start (A-225)
if m.initial[viewstate.ReachBucket] {
return
}
if r, err := viewstate.DecodeReach(e.Value); err == nil && !e.Deleted {
m.kept[e.Key] = r
}
case AnswersBucket:
if e.Deleted {
delete(m.answers, e.Key)
delete(m.answerOf, e.Key)
return
}
// The answer's warrant is `<router id>@<time>`: it ends that asking, and only that one (a later asking of the key is
// still open). An answer that cannot be read ends none.
id := ""
if a, err := meshdeliveryclient.DecodeAnswer(e.Value); err == nil {
id, _, _ = strings.Cut(a.Warrant, "@")
m.answerOf[e.Key] = a
} else {
delete(m.answerOf, e.Key)
}
m.answers[e.Key] = id
case builds.BuilderBuildsBucket:
if e.Deleted {
delete(m.keys, e.Key)
return
}
b, err := builds.DecodeBuilderBuild(e.Value)
if err != nil {
m.keys[e.Key] = keyEntry{err: err.Error()}
return
}
m.keys[e.Key] = keyEntry{value: b}
}
}
func errText(err error) string {
if err == nil {
return ""
}
return err.Error()
}
// Snapshot derives one row per module and node with an open assignment, a row marked left behind for each have with
// none, and a row per node's own key. Rows are ordered by node and module.
func (m *Model) Snapshot() Snapshot {
s := m.snapshot()
m.mu.Lock()
b := m.backlog
m.mu.Unlock()
s.Backlog = []BacklogRow{}
if b != nil {
s.Backlog = b.Rows() // read from the bus outside the model's lock
}
return s
}
// SetBacklog gives the model the seats' workers to read the backlog of asks from.
func (m *Model) SetBacklog(b *Backlog) {
m.mu.Lock()
defer m.mu.Unlock()
m.backlog = b
}
func (m *Model) snapshot() Snapshot {
m.mu.Lock()
defer m.mu.Unlock()
rows := []Row{}
for _, a := range m.placedPairs() {
r := Row{Node: a.Node, Module: a.Module, Status: StatusAssigned}
var waits []string
hk, herr := nodeengineclient.HaveKey(a.Node, a.Module)
if herr != nil {
continue
}
h, has := m.haves[hk]
// The controller no longer writes the build of an assignment (R6): the row is the want and the have of the node.
w, wanted := m.wants[hk]
wanted = wanted && w.err == "" && !w.value.Removed && w.value.Fp != ""
if wanted {
r.Assigned = w.value.Fp
}
switch {
case !has:
waits = append(waits, "no have yet")
case h.err != "":
r.Since, r.Error = h.at, h.err
default:
r.Have, r.Phase, r.Health = h.have.Build, string(h.have.Phase), string(h.have.Health)
if r.Have == "" {
r.Have = h.have.Fp
}
r.Since, r.Detail = h.at, h.have.Detail
if wanted && h.have.Fp != w.value.Fp {
waits = append(waits, "the node has not applied "+w.value.Fp)
}
}
r.Waits = strings.Join(waits, "; ")
rows = append(rows, r)
}
for key, h := range m.haves {
node, module, _, perr := nodeengineclient.ParseHaveKey(key)
if perr != nil {
continue
}
if _, ok := m.placed.Since(node, module); ok {
continue
}
r := Row{Node: node, Module: module, Status: StatusLeftBehind, Since: h.at, Error: h.err}
if h.err == "" && h.have.Phase == nodeengineclient.PhaseRemoved {
// The node-engine says it took the module away: plainly removed, not left behind.
r.Status = StatusRemoved
}
if h.err == "" {
r.Have, r.Phase, r.Health, r.Detail = h.have.Build, string(h.have.Phase), string(h.have.Health), h.have.Detail
if r.Have == "" {
r.Have = h.have.Fp
}
}
rows = append(rows, r)
}
sort.Slice(rows, func(i, j int) bool {
if rows[i].Node != rows[j].Node {
return rows[i].Node < rows[j].Node
}
return rows[i].Module < rows[j].Module
})
nodes := []NodeRow{}
for name, n := range m.nodes {
nodes = append(nodes, NodeRow{Node: name, Windows: len(n.node.Windows), Busy: len(n.node.Busy),
Strays: len(n.node.Strays), Since: n.at, Error: n.err})
}
sort.Slice(nodes, func(i, j int) bool { return nodes[i].Node < nodes[j].Node })
offline, stuck, unreadable := []OfflineRow{}, []StuckRow{}, []UnreadableRow{}
for key, p := range m.presence {
if p.err != "" {
unreadable = append(unreadable, UnreadableRow{Bucket: PresenceBucket, Key: key, Error: p.err})
continue
}
if p.value.Online {
continue
}
node, part, instance, err := meshwatchdogsclient.ParsePresenceKey(key)
if err != nil {
unreadable = append(unreadable, UnreadableRow{Bucket: PresenceBucket, Key: key, Error: err.Error()})
continue
}
if part != "" {
offline = append(offline, OfflineRow{Kind: "part", Part: part, Instance: instance, Since: p.at})
} else {
offline = append(offline, OfflineRow{Kind: "node", Node: node, Since: p.at})
}
}
for key, st := range m.stuck {
if st.err != "" {
unreadable = append(unreadable, UnreadableRow{Bucket: StuckBucket, Key: key, Error: st.err})
continue
}
if st.value.Moved {
continue
}
stuck = append(stuck, StuckRow{Bucket: st.value.Bucket, Key: st.value.Key, Waited: st.value.Waited, Since: st.value.Since})
}
for key, g := range m.groups {
if g.err != "" {
unreadable = append(unreadable, UnreadableRow{Bucket: CandidatesBucket, Key: key, Error: g.err})
}
}
for key, mod := range m.modules {
if mod.err != "" {
unreadable = append(unreadable, UnreadableRow{Bucket: ModulesBucket, Key: key, Error: mod.err})
}
}
for key, a := range m.aasks {
if a.err != "" {
unreadable = append(unreadable, UnreadableRow{Bucket: AssignerAsksBucket, Key: key, Error: a.err})
}
}
for key, w := range m.wants {
if w.err != "" {
unreadable = append(unreadable, UnreadableRow{Bucket: AssignerWantsBucket, Key: key, Error: w.err})
}
}
sort.Slice(offline, func(i, j int) bool {
a, b := offline[i], offline[j]
return a.Kind+a.Node+a.Part+a.Instance < b.Kind+b.Node+b.Part+b.Instance
})
sort.Slice(stuck, func(i, j int) bool {
return stuck[i].Bucket+"."+stuck[i].Key < stuck[j].Bucket+"."+stuck[j].Key
})
sort.Slice(unreadable, func(i, j int) bool {
return unreadable[i].Bucket+"."+unreadable[i].Key < unreadable[j].Bucket+"."+unreadable[j].Key
})
return Snapshot{Rows: rows, Nodes: nodes, Offline: offline, Stuck: stuck, Unreadable: unreadable,
Groups: m.groupRows(), Fingerprints: m.fingerprintRows(), Asks: m.askRows(), Wants: m.wantRows(), Sources: m.sourceRows()}
}
// groupRows are the readable candidate groups, ordered by group, their pull requests by repository and number.
func (m *Model) groupRows() []GroupRow {
out := []GroupRow{}
for key, g := range m.groups {
if g.err != "" {
continue
}
row := GroupRow{Group: key, Pulls: []GroupPull{}}
for _, p := range g.value.Pulls {
head := p.Head
if len(head) > 12 {
head = head[:12]
}
row.Pulls = append(row.Pulls, GroupPull{Repository: p.Repository, Number: int(p.Number), Head: head, Modules: append([]string{}, p.Modules...)})
}
sort.Slice(row.Pulls, func(i, j int) bool {
a, b := row.Pulls[i], row.Pulls[j]
return a.Repository < b.Repository || (a.Repository == b.Repository && a.Number < b.Number)
})
out = append(out, row)
}
sort.Slice(out, func(i, j int) bool { return out[i].Group < out[j].Group })
return out
}
// fingerprintRows compare each module the indexer fingerprinted with the source fingerprint of the module's newest
// successful trunk build (buildrecord.BuildRecord.IsTrunkBuild: an outcome, not failed, on main or with no ref) that recorded one. The newest key is any state and ref, so it is not
// used: the records are.
func (m *Model) fingerprintRows() []FingerprintRow {
out := []FingerprintRow{}
for name, mod := range m.modules {
if mod.err != "" {
continue
}
row := FingerprintRow{Module: name, Repository: mod.value.Repository, Commit: mod.value.Commit}
if mod.value.Fp2 == "" && mod.value.None2 == "" {
// The builder builds the new form alone, and its done keys are the build history the view reads (ADR 0571 §1).
row.State, row.Reason = FPNone, mod.value.None
if row.Reason == "" {
row.Reason = "the indexer holds no fingerprint of the new form for it, so no build of the builder's is compared"
}
out = append(out, row)
continue
}
out = append(out, m.newFormRow(name, row, mod.value))
}
sort.Slice(out, func(i, j int) bool { return out[i].Module < out[j].Module })
return out
}
// newFormRow is a module's row of the new form: none with its reason, or its fp2 against the state of the builder's key
// of it (I5, issue 1851; the build records name no fingerprint of the new form).
func (m *Model) newFormRow(name string, row FingerprintRow, v meshcatalog.ModulesByModuleValue) FingerprintRow {
if v.Fp2 == "" {
row.State, row.Reason = FPNone, v.None2
return row
}
row.Indexer = v.Fp2
key, err := builds.BuilderBuildKey(name, v.Fp2)
if err != nil {
row.State, row.Reason = FPNoKey, err.Error()
return row
}
k, ok := m.keys[key]
switch {
case !ok:
row.State = FPNoKey
case k.err != "":
row.State, row.Reason = FPBuildFailed, "the key cannot be read: "+k.err
case k.value.State == builds.BuildDone:
row.State, row.Build = FPBuilt, v.Fp2
case k.value.State == builds.BuildFailed:
row.State, row.Reason = FPBuildFailed, k.value.Error
default:
row.State, row.Reason = FPBuilding, strings.TrimSpace(k.value.State+" "+k.value.Waits)
}
return row
}
// Counted is the number of rows that are modules on the nodes, not left behind.
func (s Snapshot) Counted() int {
n := 0
for _, r := range s.Rows {
if r.Status == StatusAssigned {
n++
}
}
return n
}
// The statuses of a row.
const (
StatusAssigned = "assigned"
StatusLeftBehind = "left behind"
StatusRemoved = "removed"
)
// askRows are the readable asks with no answer under their key, ordered by key: each with its headline, its words, the
// pull requests it covers (repository, number, head shortened to 12 digits) and how to answer.
func (m *Model) askRows() []AskRow {
out := []AskRow{}
for key, a := range m.aasks {
if a.err != "" || m.answers[key] == a.ask.ID {
continue
}
row := AskRow{Key: key, Headline: a.ask.Headline, Words: a.ask.Explanation, Expires: expiry(a.ask), How: AnswerInTheView, Pulls: []string{},
Chosen: m.chosenOf(key, a.ask)}
for _, o := range a.ask.Options {
row.Options = append(row.Options, o.ID)
}
out = append(out, row)
}
sort.Slice(out, func(i, j int) bool { return out[i].Key < out[j].Key })
return out
}
// expiry is when the ask lapses; zero when it says none or says it unreadably.
func expiry(a oc.Ask) time.Time {
t, err := occ.Time(a.Expires)
if err != nil {
return time.Time{}
}
return t
}
// decodeAssignerAsk reads a key of the assigner's asks (mesh-sdk go/delivery DecodeAssignerAsk): the router's ask under its
// asking's id, offering an option (occ.Carried).
func decodeAssignerAsk(key string, value []byte) (oc.Ask, error) {
a, err := asks.DecodeAsk(key, value)
if err != nil {
return oc.Ask{}, fmt.Errorf("the assigner's ask %s is not readable: %w", key, err)
}
ask, err := occ.Carried(a.Ask, meshdeliveryclient.AskingID(key, a.Asking))
if err != nil {
return oc.Ask{}, fmt.Errorf("the assigner's ask %s is not readable: %w", key, err)
}
return ask, nil
}
// chosenOf is the option the view wrote for this asking of the ask: an answer to another asking (a stale one) is none.
func (m *Model) chosenOf(key string, ask oc.Ask) string {
if a, ok := m.chosen[key]; ok && a.Asking == ask.ID {
return a.Option
}
return ""
}
// Writer puts a key of the view's bucket answers, the only state the view writes.
type Writer interface {
Put(key string, value []byte) error
}
// SetWriter gives the model the bucket its answers are written to.
func (m *Model) SetWriter(w Writer) {
m.mu.Lock()
defer m.mu.Unlock()
m.writer = w
}
// Answer writes the view's answer to the open ask at askKey: option, one the ask offers, and the operator's words. It writes
// `<ask key>.view` and nothing else; an ask that needs proof of who answered (the merge's Approve, issue 953) is refused with
// its reason, as is an ask the view does not hold open, an option the ask does not offer, and one the view answered already.
func (m *Model) Answer(askKey, option, words string) error {
key, err := viewstate.ViewAnswerKey(askKey)
if err != nil {
return err
}
option = strings.TrimSpace(option)
if option == "" {
return fmt.Errorf("an answer names no option")
}
m.mu.Lock()
a, held := m.aasks[askKey]
open := held && a.err == "" && m.answers[askKey] != a.ask.ID
chosen, w := m.chosenOf(askKey, a.ask), m.writer
m.mu.Unlock()
if !open {
return fmt.Errorf("the view holds no open ask %s", askKey)
}
if !slices.ContainsFunc(a.ask.Options, func(o oc.Option) bool { return o.ID == option }) {
ids := []string{}
for _, o := range a.ask.Options {
ids = append(ids, o.ID)
}
return fmt.Errorf("the ask %s offers no option %q (it offers %s)", askKey, option, strings.Join(ids, ", "))
}
if chosen != "" {
return fmt.Errorf("the view answered %s with %q already: the teller fills the ask from it", askKey, chosen)
}
if w == nil {
return fmt.Errorf("the view cannot write its answers here")
}
body, err := viewstate.EncodeViewAnswer(viewstate.ViewAnswer{Asking: a.ask.ID, Option: option, Words: strings.TrimSpace(words)})
if err != nil {
return err
}
return w.Put(key, body)
}
// wantRows are the readable wants beside the haves of their nodes, ordered by node and module.
func (m *Model) wantRows() []WantRow {
out := []WantRow{}
for key, w := range m.wants {
if w.err != "" {
continue
}
node, module := splitKey(key)
v := w.value
row := WantRow{Node: node, Module: module, Want: v.Fp, Newest: v.Newest, Rule: int(v.Rule), Words: v.Words}
h, has := m.haves[key]
if has && h.err == "" {
row.Have, row.Phase, row.Health = h.have.Fp, string(h.have.Phase), string(h.have.Health)
}
if has && h.err == "" && h.have.Phase == nodeengineclient.PhaseWaitingYes && !v.Removed {
row.Judgement, row.Judged = string(h.have.Judgement), h.have.Detail
}
switch {
case v.Removed:
row.State = WantRemoved
case v.Rule != 0 || v.Newest != "":
row.State = WantHeldBack
case row.Have == v.Fp && h.have.Health == nodeengineclient.HealthFailed:
row.State = WantFailed
case row.Have == v.Fp && h.have.Phase == nodeengineclient.PhaseRunning && h.have.Health == nodeengineclient.HealthOK:
row.State = WantApplied
default:
row.State = WantPending
}
out = append(out, row)
}
sort.Slice(out, func(i, j int) bool {
if out[i].Node != out[j].Node {
return out[i].Node < out[j].Node
}
return out[i].Module < out[j].Module
})
return out
}
// sourceRows are the rows of every watched branch's readable source, in branch order (design 47 G A-173).
func (m *Model) sourceRows() []SourceRow {
var out []SourceRow
for _, b := range slices.Sorted(maps.Keys(m.sources)) {
if row := m.sourceRow(b); row != nil {
out = append(out, *row)
}
}
return out
}
// sourceRow derives the state of one watched branch's source revision (ADR 0429 §1; recorded by no part), branch escaped:
// delivered when every want of a module whose branch it is equals its module's fingerprint and every have equals its want,
// running and healthy, and every node such a module is assigned to has a want, compared only on the nodes present: a node the
// watchman says is offline is named away and left out (ADR 0603 §7, changing ADR 0429 §1). It is nil when the gate's verb
// answers no readable source of the branch.
func (m *Model) sourceRow(branch string) *SourceRow {
src := m.sources[branch]
if src == nil || src.err != "" {
return nil
}
waiting := []string{}
short := func(fp string) string {
fp = strings.TrimPrefix(fp, meshcatalogclient.SourcePrefix2)
if len(fp) > 12 {
fp = fp[:12]
}
return fp
}
looked := 0
away := m.awayNodes()
for key, w := range m.wants {
node, module := splitKey(key)
if away[node] {
continue
}
mod, ok := m.modules[module]
if w.err != "" || w.value.Removed || !ok || mod.err != "" || mod.value.Fp2 == "" || m.modBranch[module] != branch {
continue
}
looked++
h := m.haves[key]
switch {
case w.value.Fp != mod.value.Fp2:
waiting = append(waiting, fmt.Sprintf("%s on %s wants %s, not the module's %s", module, node, short(w.value.Fp), short(mod.value.Fp2)))
case h.err != "" || h.have.Fp != w.value.Fp:
waiting = append(waiting, fmt.Sprintf("%s on %s does not yet run %s", module, node, short(w.value.Fp)))
case h.have.Health == nodeengineclient.HealthFailed:
waiting = append(waiting, fmt.Sprintf("%s on %s runs %s and its health check failed", module, node, short(w.value.Fp)))
case h.have.Phase != nodeengineclient.PhaseRunning || h.have.Health != nodeengineclient.HealthOK:
waiting = append(waiting, fmt.Sprintf("%s on %s runs %s, not yet running and healthy", module, node, short(w.value.Fp)))
}
}
for _, a := range m.placedPairs() {
wk, werr := nodeengineclient.HaveKey(a.Node, a.Module)
if werr != nil || away[a.Node] {
continue
}
if _, ok := m.wants[wk]; ok {
continue
}
if mod, ok := m.modules[a.Module]; ok && mod.err == "" && mod.value.Fp2 != "" && m.modBranch[a.Module] == branch {
waiting = append(waiting, fmt.Sprintf("%s on %s has no want yet", a.Module, a.Node))
}
}
sort.Strings(waiting)
row := &SourceRow{Branch: src.branch, Revision: src.revision, State: SourceDelivered, Wants: looked, Away: slices.Sorted(maps.Keys(away))}
if len(waiting) > 0 {
row.State, row.Waiting = SourceNotDelivered, waiting
}
return row
}
// awayNodes are the nodes whose presence the watchman says is offline (a node's own key, not a part instance's): a source
// revision is delivered over the nodes present (ADR 0603 §7). Called with the lock held.
func (m *Model) awayNodes() map[string]bool {
out := map[string]bool{}
for key, p := range m.presence {
if p.err != "" || p.value.Online {
continue
}
if node, part, _, err := meshwatchdogsclient.ParsePresenceKey(key); err == nil && part == "" {
out[node] = true
}
}
return out
}
// splitKey is the node and the module (unescaped, issue 1150) of a model key, which is the have bucket's key form.
func splitKey(key string) (node, module string) {
node, module, isNode, err := nodeengineclient.ParseHaveKey(key)
if err != nil || isNode {
node, module, _ = strings.Cut(key, ".")
}
return node, module
}
// The events of a module in motion (issue 1906 D10, for the messenger, ADR 0308): with no plan to say it, the view says that a
// module's want on a node differs from its have (moving), that the have reached the want running and healthy (settled), or that it
// is held or failed or stopped moving past the watchman's bound (stuck). Each is said once for a node, a module and a composition.
const (
MotionMoving = motion.MotionMoving
MotionSettled = motion.MotionSettled
MotionStuck = motion.MotionStuck
// MotionBoundSeconds is how long a node may take to apply a want before the watchman says it stuck (watch.StepBound).
MotionBoundSeconds = 1800
)
// MotionBody is the body of the three events. Outcome and Words are those of settled and stuck: settled is "running"; stuck is
// "failed" (a health check failed) or "stopped" (the module is held, or the watchman says the want stopped moving past its bound),
// with the have's own words or the watchman's in Words.
type MotionBody struct {
Node string `json:"node"`
Module string `json:"module"`
FromFP string `json:"from_fp"`
ToFP string `json:"to_fp"`
Composed string `json:"composed"`
Since time.Time `json:"since"`
BoundSeconds int `json:"bound_seconds"`
// At is when the have or the want last changed, the time the event is about.
At time.Time `json:"at"`
Outcome string `json:"outcome,omitempty"`
Words string `json:"words,omitempty"`
}
// MotionEvent is an event to say: its type (moving, settled or stuck), its id (the bus's dedup id and the event id header, one for
// a type, a node, a module and a composition) and its body.
type MotionEvent struct {
Type string
ID string
Body MotionBody
}
// SetMotionEvent sets where the events of modules in motion go; nothing is said until it is set.
func (m *Model) SetMotionEvent(say func(MotionEvent) error) {
m.mu.Lock()
defer m.mu.Unlock()
m.motionEvent = say
}
// motionNow is the events to say now about the node and module the change touched: a want, a have, or the watchman's stuck key of a
// want.
func (m *Model) motionNow(e Entry) []MotionEvent {
if m.motionEvent == nil || !m.motionReady() {
return nil
}
var hk string
switch e.Bucket {
case AssignerWantsBucket:
node, module, err := meshwantsclient.ParseWantKey(e.Key)
if err != nil {
return nil
}
hk, _ = nodeengineclient.HaveKey(node, module)
case HaveBucket:
if _, _, isNode, err := nodeengineclient.ParseHaveKey(e.Key); err != nil || isNode {
return nil
}
hk = e.Key
case StuckBucket:
st, ok := m.stuck[e.Key]
if !ok || st.err != "" || st.value.Bucket != AssignerWantsBucket {
return nil
}
node, module, err := meshwantsclient.ParseWantKey(st.value.Key)
if err != nil {
return nil
}
hk, _ = nodeengineclient.HaveKey(node, module)
default:
return nil
}
return m.motionOf(hk)
}
// motionReady says whether the view has read what stands in the buckets motion is derived from (the wants, the haves and the
// watchman's stuck keys): before, a want read ahead of its have would be said moving though the node runs it.
func (m *Model) motionReady() bool {
for _, b := range []string{AssignerWantsBucket, HaveBucket, StuckBucket} {
if !m.initial[b] {
return false
}
}
return true
}
// Initial tells the model that the watch of a bucket has handed over every key that stood at its start. When the buckets motion is
// derived from have all done so, the events of the modules in motion at that moment are said (the ids deduplicate what a restart
// of the view says again).
func (m *Model) Initial(bucket string) {
m.mu.Lock()
was := m.motionReady()
m.initial[bucket] = true
var out []MotionEvent
if !was && m.motionReady() && m.motionEvent != nil {
keys := make([]string, 0, len(m.wants))
for k := range m.wants {
keys = append(keys, k)
}
sort.Strings(keys)
for _, k := range keys {
out = append(out, m.motionOf(k)...)
}
}
say := m.motionEvent
writes, reached := m.reachNow()
put, sayDelivered := m.reachPut, m.deliveredEvent
m.mu.Unlock()
m.sayMotion(say, out)
m.sendReach(put, sayDelivered, writes, reached)
}
// motionOf is the events to say now about the node and module of a model key, from their want, have and stuck key.
func (m *Model) motionOf(hk string) []MotionEvent {
w, ok := m.wants[hk]
if !ok || w.err != "" || w.value.Removed || w.value.Fp == "" {
return nil
}
node, module := splitKey(hk)
h, has := m.haves[hk]
has = has && h.err == ""
body := MotionBody{Node: node, Module: module, ToFP: w.value.Fp, Composed: w.value.Composed, Since: w.at, BoundSeconds: MotionBoundSeconds}
body.At = w.at
if has {
body.FromFP = h.have.Fp
if h.at.After(body.At) {
body.At = h.at
}
}
id := func(kind string) string {
return kind + "." + node + "." + module + "." + strings.TrimPrefix(w.value.Composed, "sha256:")
}
var out []MotionEvent
say := func(kind string, b MotionBody) {
if m.told[id(kind)] {
return
}
m.told[id(kind)] = true
out = append(out, MotionEvent{Type: kind, ID: id(kind), Body: b})
}
// The derivation is mesh-sdk's (motion.MotionOf, issue 1952 S7), shared with the condition keeper's hold.
words, stuck := m.stuckWant(w.value, node, module)
mo := motionOf(w.value, true, h.have, has, words, stuck)
if mo.State == motion.MotionSettled {
if m.told[id(MotionMoving)] {
b := body
b.Outcome = mo.Outcome
say(MotionSettled, b)
}
return out
}
if !m.told[id(MotionMoving)] {
say(MotionMoving, body)
}
if mo.State == motion.MotionStuck {
stop := body
stop.Outcome, stop.Words = mo.Outcome, mo.Words
say(MotionStuck, stop)
}
return out
}
// stuckWant is the watchman's words for the want of module on node, when it says the want stopped moving and not yet moved.
func (m *Model) stuckWant(_ meshwantsclient.Want, node, module string) (string, bool) {
wk, err := meshwantsclient.WantKey(node, module)
if err != nil {
return "", false
}
sk, err := meshwatchdogsclient.StuckKey(AssignerWantsBucket, wk)
if err != nil {
return "", false
}
if st, ok := m.stuck[sk]; ok && st.err == "" && !st.value.Moved {
return st.value.Waited, true
}
return "", false
}