diff --git a/modules/mesh-delivery-teller/go.mod b/modules/mesh-delivery-teller/go.mod index 7138f28..f93f07f 100644 --- a/modules/mesh-delivery-teller/go.mod +++ b/modules/mesh-delivery-teller/go.mod @@ -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 ) diff --git a/modules/mesh-delivery-view/cmd/mesh-delivery-view/integration_test.go b/modules/mesh-delivery-view/cmd/mesh-delivery-view/integration_test.go index 52d601e..3be7372 100644 --- a/modules/mesh-delivery-view/cmd/mesh-delivery-view/integration_test.go +++ b/modules/mesh-delivery-view/cmd/mesh-delivery-view/integration_test.go @@ -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") +} diff --git a/modules/mesh-delivery-view/cmd/mesh-delivery-view/main.go b/modules/mesh-delivery-view/cmd/mesh-delivery-view/main.go index 4452978..e3f5faa 100644 --- a/modules/mesh-delivery-view/cmd/mesh-delivery-view/main.go +++ b/modules/mesh-delivery-view/cmd/mesh-delivery-view/main.go @@ -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 { diff --git a/modules/mesh-delivery-view/go.mod b/modules/mesh-delivery-view/go.mod index 1d89b48..0270e9a 100644 --- a/modules/mesh-delivery-view/go.mod +++ b/modules/mesh-delivery-view/go.mod @@ -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 ) diff --git a/modules/mesh-delivery-view/internal/view/differences_test.go b/modules/mesh-delivery-view/internal/view/differences_test.go new file mode 100644 index 0000000..b8fa993 --- /dev/null +++ b/modules/mesh-delivery-view/internal/view/differences_test.go @@ -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()) + } +} diff --git a/modules/mesh-delivery-view/internal/view/serve.go b/modules/mesh-delivery-view/internal/view/serve.go index d36b09d..883fdd7 100644 --- a/modules/mesh-delivery-view/internal/view/serve.go +++ b/modules/mesh-delivery-view/internal/view/serve.go @@ -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(`{{range .}}{{.}}{{end}}{{end}} {{if .Groups}}

CANDIDATES

{{range .Groups}}{{range .}}{{end}}{{end}}
GROUPPULL REQUESTHEADMODULES
{{.}}
{{end}} {{if .Asks}}

ASKS

{{range .Asks}}{{range .}}{{end}}{{end}}
ASKHEADLINEWORDSPULL REQUESTSANSWER
{{.}}
{{end}} +{{if .Differences}}

WANTS AGAINST SENDS

{{range .DifferenceHead}}{{end}}{{range .Differences}}{{range .}}{{end}}{{end}}
{{.}}
{{.}}
{{end}} {{if .Fingerprints}}

FINGERPRINTS

{{range .Fingerprints}}{{range .}}{{end}}{{end}}
MODULESTATEINDEXER AGAINST BUILD
{{.}}
{{end}} {{if .Offline}}

OFFLINE

{{range .Offline}}{{range .}}{{end}}{{end}}
{{.}}
{{end}} {{if .Stuck}}

STUCK

{{range .Stuck}}{{range .}}{{end}}{{end}}
BUCKETKEYWAITED FORSINCE
{{.}}
{{end}} {{if .Unreadable}}

UNREADABLE

{{range .Unreadable}}{{range .}}{{end}}{{end}}
{{.}}
{{end}}`)) +var wantsPage = template.Must(template.New("w").Parse(` +Wants against sends +

What each node runs

WANTS AGAINST SENDS

+{{if .Differences}}{{range .Head}}{{end}}{{range .Differences}}{{range .}}{{end}}{{end}}
{{.}}
{{.}}
+{{else}}

no difference: every want is what the controller sends.

{{end}}`)) + // 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}) }) } diff --git a/modules/mesh-delivery-view/internal/view/view.go b/modules/mesh-delivery-view/internal/view/view.go index 2cebc2e..d884248 100644 --- a/modules/mesh-delivery-view/internal/view/view.go +++ b/modules/mesh-delivery-view/internal/view/view.go @@ -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 . asks map[string]askRecord // the gate's asks, by key + diffs map[string]diffEntry // the assigner's differences, by . 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 +} diff --git a/modules/mesh-delivery-view/manifest_test.go b/modules/mesh-delivery-view/manifest_test.go index b3c0e1c..e690a08 100644 --- a/modules/mesh-delivery-view/manifest_test.go +++ b/modules/mesh-delivery-view/manifest_test.go @@ -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) { diff --git a/modules/mesh-delivery-view/module.json b/modules/mesh-delivery-view/module.json index 09cd55b..c9d8718 100644 --- a/modules/mesh-delivery-view/module.json +++ b/modules/mesh-delivery-view/module.json @@ -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",