Files
mesh-delivery/modules/mesh-delivery-view/internal/view/view.go
T
jschoubben 9bd84941ee Show a fingerprint of the new form against the builder's key of it, since the indexer no longer writes the old form and no build record names the new (issue 1851, I5)
The view watches the builder's builds; a value written before R4e keeps today's comparison. Pins mesh-sdk feat/r4e-sdk
9cb4d59.
2026-10-22 18:16:18 +02:00

661 lines
21 KiB
Go

// Package view derives the operator's rows from the delivery buckets.
package view
import (
"encoding/json"
"fmt"
"sort"
"strings"
"sync"
"time"
"git.novox.be/novox/mesh-sdk/go/delivery"
)
// 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
}
// 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"`
}
// 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"`
}
// AnswerOnThePhone is what an open ask says about answering it: the view only shows, and the Approve is answered on the
// operator's phone (ADR 0437 §10: answering in the view is release 5's).
const AnswerOnThePhone = "answer on the phone"
// 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
assigns map[string]assignment
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>
asks map[string]askRecord // the gate's asks, by key
answers map[string]string // by key: the router id of the asking the teller's answer is for ("" when unreadable)
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
}
type presenceEntry struct {
value delivery.Presence
err string
at time.Time
}
type stuckEntry struct {
value delivery.Stuck
err string
}
// assignment holds only the fields of the controller's Assignment the view reads.
type assignment struct {
Node string `json:"node"`
Module string `json:"module"`
Build string `json:"build"`
Commit string `json:"commit"`
Since time.Time `json:"since"`
Ended time.Time `json:"ended"`
State string `json:"state"`
PulledIn bool `json:"pulled-in"`
RequiredBy string `json:"required-by"`
}
// open says whether the value is an assignment that stands for the module on its node, as the controller's
// Assignment.Stands does. A module pulled in by another stands too.
func (a assignment) open() bool { return a.Ended.IsZero() && a.State == "" }
// 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 delivery.BuildRecord
}
type groupEntry struct {
value delivery.Candidate
err string
}
type keyEntry struct {
value delivery.BuilderBuild
err string
}
type moduleEntry struct {
value delivery.Module
err string
}
type askRecord struct {
value delivery.Ask
err string
}
type haveEntry struct {
have delivery.Have
err string
at time.Time
}
type nodeEntry struct {
node delivery.HaveNode
err string
at time.Time
}
// New is an empty model.
func New() *Model {
return &Model{assigns: map[string]assignment{}, builds: map[string]buildEntry{}, groups: map[string]groupEntry{}, modules: map[string]moduleEntry{}, keys: map[string]keyEntry{}, asks: map[string]askRecord{}, answers: map[string]string{}, newest: map[string]string{},
haves: map[string]haveEntry{}, nodes: map[string]nodeEntry{},
presence: map[string]presenceEntry{}, stuck: map[string]stuckEntry{}}
}
// Apply takes one change of a key. A value it cannot read is kept as an error on its row, never guessed.
func (m *Model) Apply(e Entry) {
m.mu.Lock()
defer m.mu.Unlock()
switch e.Bucket {
case delivery.HaveBucket:
node, module, isNode, err := delivery.ParseHaveKey(e.Key)
if err != nil {
return
}
if isNode {
if e.Deleted {
delete(m.nodes, node)
return
}
n, err := delivery.DecodeHaveNode(e.Value)
m.nodes[node] = nodeEntry{node: n, err: errText(err), at: e.At}
return
}
key := node + "." + module
if e.Deleted {
delete(m.haves, key)
return
}
h, err := delivery.DecodeHave(e.Value)
m.haves[key] = haveEntry{have: h, err: errText(err), at: e.At}
case delivery.PresenceBucket:
if e.Deleted {
delete(m.presence, e.Key)
return
}
p, err := delivery.DecodePresence(e.Value)
m.presence[e.Key] = presenceEntry{value: p, err: errText(err), at: e.At}
case delivery.StuckBucket:
if e.Deleted {
delete(m.stuck, e.Key)
return
}
st, err := delivery.DecodeStuck(e.Value)
m.stuck[e.Key] = stuckEntry{value: st, err: errText(err)}
case AssignmentsBucket:
if e.Deleted {
delete(m.assigns, e.Key)
return
}
var a assignment
if json.Unmarshal(e.Value, &a) != nil || a.Node == "" || a.Module == "" {
return
}
m.assigns[a.Node+"."+a.Module] = a
case delivery.CandidatesBucket:
if e.Deleted {
delete(m.groups, e.Key)
return
}
c, err := delivery.DecodeCandidate(e.Value)
m.groups[e.Key] = groupEntry{value: c, err: errText(err)}
case delivery.ModulesBucket:
if e.Deleted {
delete(m.modules, e.Key)
return
}
mod, err := delivery.DecodeModule(e.Value)
m.modules[e.Key] = moduleEntry{value: mod, err: errText(err)}
case delivery.AsksBucket:
if e.Deleted {
delete(m.asks, e.Key)
return
}
a, err := delivery.DecodeAsk(e.Value)
m.asks[e.Key] = askRecord{value: a, err: errText(err)}
case delivery.AnswersBucket:
if e.Deleted {
delete(m.answers, 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 := delivery.DecodeAnswer(e.Value); err == nil {
id, _, _ = strings.Cut(a.Warrant, "@")
}
m.answers[e.Key] = id
case delivery.BuilderBuildsBucket:
if e.Deleted {
delete(m.keys, e.Key)
return
}
b, err := delivery.DecodeBuilderBuild(e.Value)
if err != nil {
m.keys[e.Key] = keyEntry{err: err.Error()}
return
}
m.keys[e.Key] = keyEntry{value: b}
case BuildsBucket:
module, rest, ok := strings.Cut(e.Key, ".")
if !ok {
return
}
switch {
case rest == "newest":
if e.Deleted {
delete(m.newest, module)
return
}
m.newest[module] = strings.TrimSpace(string(e.Value))
case strings.HasPrefix(rest, "build."):
if e.Deleted {
delete(m.builds, e.Key)
return
}
var b build
if json.Unmarshal(e.Value, &b) != nil {
return
}
// The source fingerprint is read with the SDK's record, not with the view's own struct.
var rec delivery.BuildRecord
if json.Unmarshal(e.Value, &rec) != nil {
return
}
m.builds[e.Key] = buildEntry{build: b, rec: rec}
}
}
}
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 {
m.mu.Lock()
defer m.mu.Unlock()
rows := []Row{}
for key, a := range m.assigns {
if !a.open() {
continue
}
r := Row{Node: a.Node, Module: a.Module, Status: StatusAssigned, Assigned: a.Build}
if a.PulledIn {
r.RequiredBy = a.RequiredBy
}
var waits []string
h, has := m.haves[key]
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)
r.Since, r.Detail = h.at, h.have.Detail
if a.Build != "" && h.have.Build != a.Build {
waits = append(waits, "the node has not applied "+a.Build)
}
}
if id := m.newest[a.Module]; id != "" && a.Build != "" && id != a.Build {
if b, ok := m.builds[a.Module+".build."+id]; ok && b.done() {
waits = append(waits, a.Module+" has a newer build "+id+" not yet assigned")
}
}
r.Waits = strings.Join(waits, "; ")
rows = append(rows, r)
}
for key, h := range m.haves {
if a, ok := m.assigns[key]; ok && a.open() {
continue
}
node, module, _, _ := delivery.ParseHaveKey(key)
r := Row{Node: node, Module: module, Status: StatusLeftBehind, Since: h.at, Error: h.err}
if h.err == "" && h.have.Phase == delivery.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
}
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: count(n.node.Windows), Busy: count(n.node.Busy),
Strays: count(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: delivery.PresenceBucket, Key: key, Error: p.err})
continue
}
if p.value.Online {
continue
}
node, part, instance, err := delivery.ParsePresenceKey(key)
if err != nil {
unreadable = append(unreadable, UnreadableRow{Bucket: delivery.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: delivery.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: delivery.CandidatesBucket, Key: key, Error: g.err})
}
}
for key, mod := range m.modules {
if mod.err != "" {
unreadable = append(unreadable, UnreadableRow{Bucket: delivery.ModulesBucket, Key: key, Error: mod.err})
}
}
for key, a := range m.asks {
if a.err != "" {
unreadable = append(unreadable, UnreadableRow{Bucket: delivery.AsksBucket, Key: key, Error: a.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()}
}
// 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.Head
if len(head) > 12 {
head = head[:12]
}
row.Pulls = append(row.Pulls, GroupPull{Repository: p.Repository, Number: 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 (delivery.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 {
newest := map[string]buildEntry{}
for _, b := range m.builds {
if b.rec.Module == "" || !b.rec.IsTrunkBuild() || b.rec.SourceFingerprint == "" {
continue
}
if old, ok := newest[b.rec.Module]; !ok || b.rec.At.After(old.rec.At) || (b.rec.At.Equal(old.rec.At) && b.rec.ID > old.rec.ID) {
newest[b.rec.Module] = b
}
}
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 new form (R4e): the indexer's fp2 against the builder's key of it.
out = append(out, m.newFormRow(name, row, mod.value))
continue
}
b, hasBuild := newest[name]
switch {
case mod.value.None != "":
row.State, row.Reason = FPNone, mod.value.None
case !hasBuild:
row.State, row.Indexer = FPNoBuild, mod.value.FP
case b.rec.SourceFingerprint == mod.value.FP:
row.State, row.Indexer, row.Build = FPSame, mod.value.FP, b.rec.SourceFingerprint
default:
row.State, row.Indexer, row.Build = FPDiffers, mod.value.FP, b.rec.SourceFingerprint
}
out = append(out, row)
}
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 delivery.Module) FingerprintRow {
if v.FP2 == "" {
row.State, row.Reason = FPNone, v.None2
return row
}
row.Indexer = v.FP2
key, err := delivery.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 == delivery.BuildDone:
row.State, row.Build = FPBuilt, v.FP2
case k.value.State == delivery.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
}
// count is the number of items of a carried JSON array or object, and 0 for anything else.
func count(raw json.RawMessage) int {
var list []json.RawMessage
if json.Unmarshal(raw, &list) == nil {
return len(list)
}
var obj map[string]json.RawMessage
if json.Unmarshal(raw, &obj) == nil {
return len(obj)
}
return 0
}
// 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 buckets the view reads besides the have bucket (which mesh-sdk's delivery package names).
const (
AssignmentsBucket = "mesh-controller_assignments"
BuildsBucket = "mesh-controller_builds"
)
// 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.asks {
if a.err != "" || m.answers[key] == a.value.Ask.ID {
continue
}
row := AskRow{Key: key, Group: a.value.Group, Headline: a.value.Ask.Headline, Words: a.value.Ask.Explanation,
Expires: a.value.Ask.Expires, How: AnswerOnThePhone, Pulls: []string{}}
for _, h := range a.value.Heads {
head := h.Head
if len(head) > 12 {
head = head[:12]
}
row.Pulls = append(row.Pulls, fmt.Sprintf("%s#%d @%s", h.Repository, h.Number, head))
}
out = append(out, row)
}
sort.Slice(out, func(i, j int) bool { return out[i].Key < out[j].Key })
return out
}