novox/mesh-delivery
19
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Show every difference between the assigner's wants and the controller's sends with its rule, so the operator sees what holds a node back (issue 1297 B8)

Browse Source
This commit is contained in:
jschoubben
2026-10-23 15:15:32 +02:00
parent 4ffe5f4d52
commit 57b0030258
9 changed files with 302 additions and 7 deletions
Download Patch File Download Diff File Expand all files Collapse all files
modules/mesh-delivery-teller/go.mod
+1 -1
View File
@@ -3,7 +3,7 @@ module mesh-delivery-teller
go 1.26.0
require (
git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261020060201-88581d0acaef
git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261023151240-d9384bcb239a
github.com/nats-io/nats-server/v2 v2.11.17
github.com/nats-io/nats.go v1.54.0
)
modules/mesh-delivery-view/cmd/mesh-delivery-view/integration_test.go
+82
View File
@@ -390,3 +390,85 @@ func TestProcessWholeShowsTheOpenAsksAndDropsTheAnsweredOnes(t *testing.T) { //
}
wait(func(s view.Snapshot) bool { return len(s.Asks) == 0 }, "the answered ask never left the list")
}
// TestProcessWholeListsEveryDifference starts the program whole before the assigner's differences bucket exists, then
// creates it and writes a difference explained by a rule and one unexplained: the socket's snapshot lists both with their
// rule, and a difference taken away leaves the list (1297/B8: the view lists every difference with its rule).
func TestProcessWholeListsEveryDifference(t *testing.T) { // 1297/B8 AC
srv, err := server.NewServer(&server.Options{Host: "127.0.0.1", Port: -1, JetStream: true, StoreDir: t.TempDir(), NoLog: true, NoSigs: true})
if err != nil {
t.Fatal(err)
}
go srv.Start()
if !srv.ReadyForConnections(10 * time.Second) {
t.Fatal("the bus did not come up")
}
t.Cleanup(srv.Shutdown)
nc, err := nats.Connect(srv.ClientURL())
if err != nil {
t.Fatal(err)
}
t.Cleanup(nc.Close)
js, err := jetstream.New(nc)
if err != nil {
t.Fatal(err)
}
ctx, cancel := context.WithTimeout(context.Background(), 60*time.Second)
defer cancel()
dir, err := os.MkdirTemp(os.Getenv("TMPDIR"), "i")
if err != nil {
t.Fatal(err)
}
t.Cleanup(func() { os.RemoveAll(dir) })
broker := filepath.Join(dir, "broker")
cred, _ := json.Marshal(credential{URL: srv.ClientURL(), Node: "n", Module: "mesh-delivery-view", User: "mesh-delivery-view", Password: "x"})
if err := os.WriteFile(broker, cred, 0o600); err != nil {
t.Fatal(err)
}
socket := filepath.Join(dir, "run", "cli.sock")
env := map[string]string{"MESH_BROKER_FILE": broker, "MESH_DELIVERY_SOCKET": socket, "MESH_DELIVERY_WEB": "127.0.0.1:0"}
runCtx, stop := context.WithCancel(ctx)
done := make(chan error, 1)
go func() { done <- run(runCtx, func(k string) string { return env[k] }, io.Discard) }()
t.Cleanup(func() { stop(); <-done })
kv, err := js.CreateKeyValue(ctx, jetstream.KeyValueConfig{Bucket: delivery.AssignerDifferencesBucket, History: delivery.AssignerDifferencesHistory})
if err != nil {
t.Fatal(err)
}
const fpOld = "src2:1111111111111111111111111111111111111111111111111111111111111111"
const fpNew = "src2:2222222222222222222222222222222222222222222222222222222222222222"
for key, d := range map[string]delivery.Difference{
"ace.plex": {Want: fpOld, Sent: fpNew, Rule: delivery.RuleCanary, Words: "ace is not the canary"},
"g14.plex": {Want: fpOld, Sent: fpNew, Unexplained: true},
} {
v, err := delivery.EncodeDifference(d)
if err != nil {
t.Fatal(err)
}
if _, err := kv.Put(ctx, key, v); err != nil {
t.Fatal(err)
}
}
wait := func(check func(view.Snapshot) bool, what string) {
t.Helper()
var last view.Snapshot
for deadline := time.Now().Add(20 * time.Second); time.Now().Before(deadline); time.Sleep(100 * time.Millisecond) {
if s, err := view.Ask(socket); err == nil {
last = s
if check(s) {
return
}
}
}
t.Fatalf("%s; last: %+v", what, last)
}
wait(func(s view.Snapshot) bool {
d := s.Differences
return len(d) == 2 && d[0].Node == "ace" && d[0].Rule == delivery.RuleCanary && d[1].Node == "g14" && d[1].Unexplained
}, "the two differences never showed with their rules")
if err := kv.Delete(ctx, "g14.plex"); err != nil {
t.Fatal(err)
}
wait(func(s view.Snapshot) bool { return len(s.Differences) == 1 }, "the difference taken away never left the list")
}
modules/mesh-delivery-view/cmd/mesh-delivery-view/main.go
+1 -1
View File
@@ -136,7 +136,7 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro
if err != nil {
return err
}
for _, bucket := range []string{delivery.HaveBucket, view.AssignmentsBucket, view.BuildsBucket, delivery.PresenceBucket, delivery.StuckBucket, delivery.CandidatesBucket, delivery.ModulesBucket, delivery.AsksBucket, delivery.AnswersBucket, delivery.BuilderBuildsBucket} {
for _, bucket := range []string{delivery.HaveBucket, view.AssignmentsBucket, view.BuildsBucket, delivery.PresenceBucket, delivery.StuckBucket, delivery.CandidatesBucket, delivery.ModulesBucket, delivery.AsksBucket, delivery.AnswersBucket, delivery.BuilderBuildsBucket, delivery.AssignerDifferencesBucket} {
go follow(ctx, js, bucket, model, logger)
}
select {
modules/mesh-delivery-view/go.mod
+1 -1
View File
@@ -3,7 +3,7 @@ module mesh-delivery-view
go 1.26.0
require (
git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261022190740-104533e8c0c9
git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261023151240-d9384bcb239a
github.com/nats-io/nats-server/v2 v2.11.17
github.com/nats-io/nats.go v1.54.0
)
modules/mesh-delivery-view/internal/view/differences_test.go
+103
View File
@@ -0,0 +1,103 @@
package view
import (
"net/http"
"net/http/httptest"
"strings"
"testing"
"git.novox.be/novox/mesh-sdk/go/delivery"
)
const (
fpOld = "src2:1111111111111111111111111111111111111111111111111111111111111111"
fpNew = "src2:2222222222222222222222222222222222222222222222222222222222222222"
)
// diffEntry is a key of the assigner's differences as the assigner writes it (mesh-sdk go/delivery EncodeDifference).
func differenceE(t *testing.T, node, module string, d delivery.Difference) Entry {
t.Helper()
key, err := delivery.DifferenceKey(node, module)
if err != nil {
t.Fatal(err)
}
v, err := delivery.EncodeDifference(d)
if err != nil {
t.Fatal(err)
}
return Entry{Bucket: delivery.AssignerDifferencesBucket, Key: key, Value: v, At: t0}
}
func TestEveryDifferenceIsListedWithItsRule(t *testing.T) { // 1297/B8 AC
m := New()
fill(m,
differenceE(t, "ace", "plex", delivery.Difference{Want: fpOld, Sent: fpNew, Rule: delivery.RuleCanary, Words: "ace is not the canary"}),
differenceE(t, "g14", "plex", delivery.Difference{Want: fpOld, Sent: fpNew, Unexplained: true}),
differenceE(t, "ace", "old", delivery.Difference{WantRemoved: true, Sent: fpOld, Rule: delivery.RuleNotHeld}),
)
got := m.Snapshot().Differences
if len(got) != 3 {
t.Fatalf("differences = %+v, want all three", got)
}
// ordered by node then module
if got[0].Node != "ace" || got[0].Module != "old" || got[1].Module != "plex" || got[2].Node != "g14" {
t.Errorf("order = %+v", got)
}
if r := got[1]; r.Want != fpOld || r.Sent != fpNew || r.Rule != delivery.RuleCanary || r.Unexplained || r.Words != "ace is not the canary" {
t.Errorf("row = %+v", r)
}
if !got[0].WantRemoved || got[0].Rule != delivery.RuleNotHeld {
t.Errorf("removed row = %+v", got[0])
}
if !got[2].Unexplained || got[2].Rule != 0 {
t.Errorf("unexplained row = %+v", got[2])
}
// a difference taken away (the want and the send agree again) leaves the list
fill(m, Entry{Bucket: delivery.AssignerDifferencesBucket, Key: "g14.plex", Deleted: true})
if got := m.Snapshot().Differences; len(got) != 2 {
t.Errorf("after the delete = %+v", got)
}
}
func TestAnUnreadableDifferenceIsNamedAndNotDropped(t *testing.T) { // never dropped silently
m := New()
fill(m, Entry{Bucket: delivery.AssignerDifferencesBucket, Key: "ace.plex", Value: []byte(`{"v":99}`), At: t0})
s := m.Snapshot()
if len(s.Differences) != 0 || len(s.Unreadable) != 1 || s.Unreadable[0].Bucket != delivery.AssignerDifferencesBucket || s.Unreadable[0].Key != "ace.plex" {
t.Errorf("snapshot = %+v", s)
}
}
func TestTableAndPageShowTheWantsAgainstTheSends(t *testing.T) { // 1297/B8 AC
m := New()
fill(m,
differenceE(t, "ace", "plex", delivery.Difference{Want: fpOld, Sent: fpNew, Rule: delivery.RuleCanary, Words: "ace is not the canary"}),
differenceE(t, "g14", "plex", delivery.Difference{Want: fpOld, Sent: fpNew, Unexplained: true}),
)
var out strings.Builder
Table(&out, m.Snapshot())
for _, w := range []string{"WANTS AGAINST SENDS", "ace", "plex", fpOld, fpNew, "rule 3", "ace is not the canary", "unexplained"} {
if !strings.Contains(out.String(), w) {
t.Errorf("table lacks %q:\n%s", w, out.String())
}
}
for _, path := range []string{"/", "/wants"} {
rec := httptest.NewRecorder()
Web(m).ServeHTTP(rec, httptest.NewRequest(http.MethodGet, path, nil))
body := rec.Body.String()
if rec.Code != 200 {
t.Fatalf("%s = %d", path, rec.Code)
}
for _, w := range []string{"WANTS AGAINST SENDS", "rule 3", "unexplained", "ace is not the canary"} {
if !strings.Contains(body, w) {
t.Errorf("%s lacks %q:\n%s", path, w, body)
}
}
}
// With none, the page says so and does not hide the heading.
rec := httptest.NewRecorder()
Web(New()).ServeHTTP(rec, httptest.NewRequest(http.MethodGet, "/wants", nil))
if !strings.Contains(rec.Body.String(), "no difference") {
t.Errorf("empty page = %q", rec.Body.String())
}
}
modules/mesh-delivery-view/internal/view/serve.go
+50 -1
View File
@@ -165,6 +165,35 @@ func askText(s Snapshot) [][]string {
return out
}
// differenceText is one line per difference: the node, the module, what the assigner wants, what the controller sends, the rule
// that explains it (or unexplained) and the assigner's words.
func differenceText(s Snapshot) [][]string {
var out [][]string
for _, d := range s.Differences {
want, sent := d.Want, d.Sent
switch {
case d.WantRemoved:
want = "removed"
case want == "":
want = "none"
}
if sent == "" {
sent = "none"
if d.SentBuild != "" {
sent = "build " + d.SentBuild
}
}
why := "unexplained"
if !d.Unexplained {
why = fmt.Sprintf("rule %d", d.Rule)
}
out = append(out, []string{d.Node, d.Module, want, sent, why, d.Words})
}
return out
}
var differenceHead = []string{"NODE", "MODULE", "WANT", "SENT", "RULE", "WORDS"}
var head = []string{"NODE", "MODULE", "ASSIGNED", "HAVE", "PHASE", "HEALTH", "SINCE", "DETAIL", "WAITS"}
// Table prints the snapshot as a table.
@@ -204,6 +233,14 @@ func Table(w io.Writer, s Snapshot) {
row(cells)
}
}
if d := differenceText(s); len(d) > 0 {
fmt.Fprintln(tw)
row([]string{"WANTS AGAINST SENDS"})
row(differenceHead)
for _, cells := range d {
row(cells)
}
}
if f := fingerprintText(s); len(f) > 0 {
fmt.Fprintln(tw)
row([]string{"MODULE", "FINGERPRINT", "INDEXER AGAINST BUILD"})
@@ -248,11 +285,18 @@ var page = template.Must(template.New("p").Parse(`<!doctype html><html><head><me
{{range .Rows}}<tr>{{range .}}<td>{{.}}</td>{{end}}</tr>{{end}}</table>
{{if .Groups}}<h2>CANDIDATES</h2><table><tr><th>GROUP</th><th>PULL REQUEST</th><th>HEAD</th><th>MODULES</th></tr>{{range .Groups}}<tr>{{range .}}<td>{{.}}</td>{{end}}</tr>{{end}}</table>{{end}}
{{if .Asks}}<h2>ASKS</h2><table><tr><th>ASK</th><th>HEADLINE</th><th>WORDS</th><th>PULL REQUESTS</th><th>ANSWER</th></tr>{{range .Asks}}<tr>{{range .}}<td>{{.}}</td>{{end}}</tr>{{end}}</table>{{end}}
{{if .Differences}}<h2>WANTS AGAINST SENDS</h2><table><tr>{{range .DifferenceHead}}<th>{{.}}</th>{{end}}</tr>{{range .Differences}}<tr>{{range .}}<td>{{.}}</td>{{end}}</tr>{{end}}</table>{{end}}
{{if .Fingerprints}}<h2>FINGERPRINTS</h2><table><tr><th>MODULE</th><th>STATE</th><th>INDEXER AGAINST BUILD</th></tr>{{range .Fingerprints}}<tr>{{range .}}<td>{{.}}</td>{{end}}</tr>{{end}}</table>{{end}}
{{if .Offline}}<h2>OFFLINE</h2><table>{{range .Offline}}<tr>{{range .}}<td>{{.}}</td>{{end}}</tr>{{end}}</table>{{end}}
{{if .Stuck}}<h2>STUCK</h2><table><tr><th>BUCKET</th><th>KEY</th><th>WAITED FOR</th><th>SINCE</th></tr>{{range .Stuck}}<tr>{{range .}}<td>{{.}}</td>{{end}}</tr>{{end}}</table>{{end}}
{{if .Unreadable}}<h2>UNREADABLE</h2><table>{{range .Unreadable}}<tr>{{range .}}<td>{{.}}</td>{{end}}</tr>{{end}}</table>{{end}}</body></html>`))
var wantsPage = template.Must(template.New("w").Parse(`<!doctype html><html><head><meta charset="utf-8"><meta http-equiv="refresh" content="5">
<title>Wants against sends</title><style>body{font:14px sans-serif;margin:1em}table{border-collapse:collapse}td,th{padding:2px 10px;text-align:left}</style></head>
<body><p><a href="/">What each node runs</a></p><h1>WANTS AGAINST SENDS</h1>
{{if .Differences}}<table><tr>{{range .Head}}<th>{{.}}</th>{{end}}</tr>{{range .Differences}}<tr>{{range .}}<td>{{.}}</td>{{end}}</tr>{{end}}</table>
{{else}}<p>no difference: every want is what the controller sends.</p>{{end}}</body></html>`))
// Web is the read-only web view of the same rows, refreshed by the browser.
func Web(m *Model) http.Handler {
return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
@@ -261,6 +305,11 @@ func Web(m *Model) http.Handler {
http.Error(w, "this view is read-only", http.StatusMethodNotAllowed)
return
}
if r.URL.Path == "/wants" {
w.Header().Set("Content-Type", "text/html; charset=utf-8")
wantsPage.Execute(w, map[string]any{"Head": differenceHead, "Differences": differenceText(m.Snapshot())})
return
}
if r.URL.Path != "/" {
http.NotFound(w, r)
return
@@ -281,6 +330,6 @@ func Web(m *Model) http.Handler {
for _, u := range snap.Unreadable {
bad = append(bad, []string{u.Bucket, u.Key, u.Error})
}
page.Execute(w, map[string]any{"Head": head, "Rows": rows, "Groups": groupText(snap), "Asks": askText(snap), "Fingerprints": fingerprintText(snap), "Offline": off, "Stuck": stuck, "Unreadable": bad})
page.Execute(w, map[string]any{"Head": head, "Rows": rows, "Groups": groupText(snap), "Asks": askText(snap), "Fingerprints": fingerprintText(snap), "Differences": differenceText(snap), "DifferenceHead": differenceHead, "Offline": off, "Stuck": stuck, "Unreadable": bad})
})
}
modules/mesh-delivery-view/internal/view/view.go
+62 -2
View File
@@ -64,6 +64,24 @@ type Snapshot struct {
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"`
// Differences is every difference between what the assigner wants a node to run and what the controller sends, with the rule that
// explains it or unexplained (novox/hq ADR 0528 clause 4, issue 1297 B8).
Differences []DifferenceRow `json:"differences"`
}
// DifferenceRow is a module on a node whose want and send differ: the fingerprint wanted (empty when the node has no want,
// WantRemoved when the want is removed), the fingerprint of the build the controller's assignment names (empty when it sends
// none; SentBuild names that build), and the rule that explains it (Rule 1 to 6) or Unexplained.
type DifferenceRow struct {
Node string `json:"node"`
Module string `json:"module"`
Want string `json:"want,omitempty"`
WantRemoved bool `json:"want-removed,omitempty"`
Sent string `json:"sent,omitempty"`
SentBuild string `json:"sent-build,omitempty"`
Rule int `json:"rule,omitempty"`
Unexplained bool `json:"unexplained,omitempty"`
Words string `json:"words,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.
@@ -155,6 +173,7 @@ type Model struct {
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
diffs map[string]diffEntry // the assigner's differences, by <node>.<module>
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
@@ -221,6 +240,11 @@ type keyEntry struct {
err string
}
type diffEntry struct {
value delivery.Difference
err string
}
type moduleEntry struct {
value delivery.Module
err string
@@ -245,7 +269,7 @@ type nodeEntry struct {
// 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{},
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{}, diffs: map[string]diffEntry{}, answers: map[string]string{}, newest: map[string]string{},
haves: map[string]haveEntry{}, nodes: map[string]nodeEntry{},
presence: map[string]presenceEntry{}, stuck: map[string]stuckEntry{}}
}
@@ -321,6 +345,13 @@ func (m *Model) Apply(e Entry) {
}
a, err := delivery.DecodeAsk(e.Value)
m.asks[e.Key] = askRecord{value: a, err: errText(err)}
case delivery.AssignerDifferencesBucket:
if e.Deleted {
delete(m.diffs, e.Key)
return
}
d, err := delivery.DecodeDifference(e.Value)
m.diffs[e.Key] = diffEntry{value: d, err: errText(err)}
case delivery.AnswersBucket:
if e.Deleted {
delete(m.answers, e.Key)
@@ -490,6 +521,11 @@ func (m *Model) Snapshot() Snapshot {
unreadable = append(unreadable, UnreadableRow{Bucket: delivery.AsksBucket, Key: key, Error: a.err})
}
}
for key, d := range m.diffs {
if d.err != "" {
unreadable = append(unreadable, UnreadableRow{Bucket: delivery.AssignerDifferencesBucket, Key: key, Error: d.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
@@ -501,7 +537,7 @@ func (m *Model) Snapshot() Snapshot {
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()}
Groups: m.groupRows(), Fingerprints: m.fingerprintRows(), Asks: m.askRows(), Differences: m.differenceRows()}
}
// groupRows are the readable candidate groups, ordered by group, their pull requests by repository and number.
@@ -658,3 +694,27 @@ func (m *Model) askRows() []AskRow {
sort.Slice(out, func(i, j int) bool { return out[i].Key < out[j].Key })
return out
}
// differenceRows are the readable differences, ordered by node and module.
func (m *Model) differenceRows() []DifferenceRow {
out := []DifferenceRow{}
for key, d := range m.diffs {
if d.err != "" {
continue
}
node, module, err := delivery.ParseDifferenceKey(key)
if err != nil {
continue
}
v := d.value
out = append(out, DifferenceRow{Node: node, Module: module, Want: v.Want, WantRemoved: v.WantRemoved, Sent: v.Sent, SentBuild: v.SentBuild,
Rule: v.Rule, Unexplained: v.Unexplained, Words: v.Words})
}
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
}
modules/mesh-delivery-view/manifest_test.go
+1 -1
View File
@@ -76,7 +76,7 @@ func TestManifestIsTakenAndReadsOnly(t *testing.T) { // 1134/AC7
if m.Module != "mesh-delivery-view" {
t.Errorf("module = %q", m.Module)
}
want := []string{"mesh-controller.assignments", "mesh-controller.builds", "mesh-delivery-builder.builds", "mesh-delivery-gate.asks", "mesh-delivery-indexer.candidates", "mesh-delivery-indexer.modules", "mesh-delivery-teller.answers", "mesh-delivery-watchman.presence", "mesh-delivery-watchman.stuck", "node-engine.have"} // 1168/AC1, 1241/AC4, 1345/AC3
want := []string{"mesh-controller.assignments", "mesh-controller.builds", "mesh-delivery-assigner.differences", "mesh-delivery-builder.builds", "mesh-delivery-gate.asks", "mesh-delivery-indexer.candidates", "mesh-delivery-indexer.modules", "mesh-delivery-teller.answers", "mesh-delivery-watchman.presence", "mesh-delivery-watchman.stuck", "node-engine.have"} // 1168/AC1, 1241/AC4, 1345/AC3, 1297/B8
got := slices.Clone(m.Reads)
slices.Sort(got)
if !reflect.DeepEqual(got, want) {
modules/mesh-delivery-view/module.json
+1
View File
@@ -14,6 +14,7 @@
"reads": [
"mesh-controller.assignments",
"mesh-controller.builds",
"mesh-delivery-assigner.differences",
"mesh-delivery-builder.builds",
"mesh-delivery-gate.asks",
"mesh-delivery-indexer.candidates",