From 57b00302581d1f8ecaca778cc9f4c4f15d82b87a Mon Sep 17 00:00:00 2001 From: jochen Date: Fri, 23 Oct 2026 15:15:32 +0200 Subject: [PATCH 1/5] 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) --- modules/mesh-delivery-teller/go.mod | 2 +- .../mesh-delivery-view/integration_test.go | 82 ++++++++++++++ .../cmd/mesh-delivery-view/main.go | 2 +- modules/mesh-delivery-view/go.mod | 2 +- .../internal/view/differences_test.go | 103 ++++++++++++++++++ .../mesh-delivery-view/internal/view/serve.go | 51 ++++++++- .../mesh-delivery-view/internal/view/view.go | 64 ++++++++++- modules/mesh-delivery-view/manifest_test.go | 2 +- modules/mesh-delivery-view/module.json | 1 + 9 files changed, 302 insertions(+), 7 deletions(-) create mode 100644 modules/mesh-delivery-view/internal/view/differences_test.go 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", From f7dd4ab586e0193afa9926626d1fbfabb8f7e8f3 Mon Sep 17 00:00:00 2001 From: jochen Date: Fri, 23 Oct 2026 15:17:40 +0200 Subject: [PATCH 2/5] Let the view answer the assigner's asks that need no proof, under the ask's view key, and refuse the merge's Approve on the phone's behalf (issue 1297 B9a) --- .../mesh-delivery-view/integration_test.go | 130 +++++++++++ .../cmd/mesh-delivery-view/main.go | 22 +- .../internal/view/answers_test.go | 217 ++++++++++++++++++ .../mesh-delivery-view/internal/view/serve.go | 123 +++++++++- .../mesh-delivery-view/internal/view/view.go | 145 +++++++++++- modules/mesh-delivery-view/manifest_test.go | 16 +- modules/mesh-delivery-view/module.json | 4 + 7 files changed, 636 insertions(+), 21 deletions(-) create mode 100644 modules/mesh-delivery-view/internal/view/answers_test.go 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 3be7372..8da74a5 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 @@ -8,6 +8,7 @@ import ( "io" "os" "path/filepath" + "strings" "testing" "time" @@ -472,3 +473,132 @@ func TestProcessWholeListsEveryDifference(t *testing.T) { // 1297/B8 AC } wait(func(s view.Snapshot) bool { return len(s.Differences) == 1 }, "the difference taken away never left the list") } + +// TestProcessWholeWritesTheAnswerToAnAskNeedingNoProof starts the program whole, with the assigner's asks, the view's +// answers and the teller's answers buckets, and a fixture hold ask (in release 5 the assigner writes none yet). The +// operator's answer over the socket is written under `.view` in the view's bucket; the ask shows the choice, and +// leaves the list once the teller's answer is under its key. The merge's Approve is refused with its reason and writes +// nothing (1297/B9a AC). +func TestProcessWholeWritesTheAnswerToAnAskNeedingNoProof(t *testing.T) { // 1297/B9a 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() + bucket := func(name string, history uint8) jetstream.KeyValue { + t.Helper() + kv, err := js.CreateKeyValue(ctx, jetstream.KeyValueConfig{Bucket: name, History: history}) + if err != nil { + t.Fatal(err) + } + return kv + } + askBucket := bucket(delivery.AssignerAsksBucket, delivery.AssignerAsksHistory) + viewBucket := bucket(delivery.ViewAnswersBucket, delivery.ViewAnswersHistory) + teller := bucket(delivery.AnswersBucket, delivery.AnswersHistory) + gate := bucket(delivery.AsksBucket, delivery.AsksHistory) + + 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 }) + + const fp = "src2:2222222222222222222222222222222222222222222222222222222222222222" + key, err := delivery.HoldAskKey("plex", fp) + if err != nil { + t.Fatal(err) + } + ask := asks.Ask{ID: delivery.AskID(key), Headline: "plex is held after a failed health check", Explanation: "ace failed the check.", Who: asks.Operator, + Expires: time.Now().Add(time.Hour), OnExpiry: "the fingerprint stays held", + Options: []asks.Option{{ID: "release", Label: "Release", Does: "lets the move go on", Level: asks.Approve}, {ID: "keep", Label: "Keep held", Does: "keeps it held", Level: asks.Approve}}} + value, _ := json.Marshal(map[string]any{"v": 1, "ask": ask}) + if _, err := askBucket.Put(ctx, key, value); 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 { return len(s.Asks) == 1 && s.Asks[0].How == view.AnswerInTheView }, "the open ask never showed") + + if err := view.AnswerAsk(socket, key, "release", "the check was a flake"); err != nil { + t.Fatal(err) + } + e, err := viewBucket.Get(ctx, key+".view") + if err != nil { + t.Fatalf("the view wrote no answer under %s.view: %v", key, err) + } + if a, err := delivery.DecodeViewAnswer(e.Value()); err != nil || a.Option != "release" { + t.Fatalf("answer = %+v, %v", a, err) + } + wait(func(s view.Snapshot) bool { return len(s.Asks) == 1 && s.Asks[0].Chosen == "release" }, "the ask never showed the view's choice") + + // The merge's Approve is answered on the phone alone. + heads := []delivery.Head{{Repository: "novox/mesh-delivery", Number: 12, Head: "0123456789abcdef0123456789abcdef01234567"}} + mkey, err := delivery.MergeAskKey("feat/x", heads) + if err != nil { + t.Fatal(err) + } + merge := asks.Ask{ID: delivery.AskID(mkey), Headline: "Merge", Explanation: "Merges.", Who: asks.Operator, Expires: time.Now().Add(time.Hour), OnExpiry: "held", + Options: []asks.Option{{ID: "approve", Label: "Approve", Does: "merges", Level: asks.Approve, Binds: "sha256:00"}}} + mv, err := delivery.EncodeAsk(delivery.Ask{Group: "feat/x", Heads: heads, Ask: merge}) + if err != nil { + t.Fatal(err) + } + if _, err := gate.Put(ctx, mkey, mv); err != nil { + t.Fatal(err) + } + wait(func(s view.Snapshot) bool { return len(s.Asks) == 2 }, "the merge ask never showed") + if err := view.AnswerAsk(socket, mkey, "approve", ""); err == nil || !strings.Contains(err.Error(), "answer it on the phone (issue 953)") { + t.Errorf("merge Approve = %v, want the refusal naming the phone", err) + } + if _, err := viewBucket.Get(ctx, mkey+".view"); err == nil { + t.Error("the view wrote an answer to the merge's ask") + } + + // The teller fills the ask: it leaves the list. + answer, err := delivery.EncodeAnswer(delivery.Answer{Option: "release", Warrant: ask.ID + "@2026-10-23T12:00:00Z", Words: "answered in the view"}) + if err != nil { + t.Fatal(err) + } + if _, err := teller.Put(ctx, key, answer); err != nil { + t.Fatal(err) + } + wait(func(s view.Snapshot) bool { return len(s.Asks) == 1 && s.Asks[0].Key == mkey }, "the answered ask 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 e3f5faa..5863ce1 100644 --- a/modules/mesh-delivery-view/cmd/mesh-delivery-view/main.go +++ b/modules/mesh-delivery-view/cmd/mesh-delivery-view/main.go @@ -1,6 +1,7 @@ // Command mesh-delivery-view is the operator's view of delivery (novox/hq ADR 0429 §3): it watches the have, assignments // and builds buckets on its own bus credential, derives one row per module and node in memory, and serves them on a -// local socket (for mesh-delivery-cli) and on a local web page. It writes nothing and starts, stops or retries nothing. +// local socket (for mesh-delivery-cli) and on a local web page. It starts, stops or retries nothing; the one thing it writes is the operator's answer to an ask that needs no proof of who +// answered, under `.view` in its bucket answers (ADR 0528 clause 6). package main import ( @@ -136,7 +137,8 @@ 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, delivery.AssignerDifferencesBucket} { + model.SetWriter(kvWriter{ctx: ctx, js: js}) + 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, delivery.AssignerAsksBucket, delivery.ViewAnswersBucket} { go follow(ctx, js, bucket, model, logger) } select { @@ -152,6 +154,22 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro return nil } +// kvWriter is the view's one write: a key of its own bucket answers (manifest state entry answers), which the teller reads. +// The bucket is the controller's to create; a bucket not there yet is an error the operator is told. +type kvWriter struct { + ctx context.Context + js jetstream.JetStream +} + +func (w kvWriter) Put(key string, value []byte) error { + kv, err := w.js.KeyValue(w.ctx, delivery.ViewAnswersBucket) + if err != nil { + return fmt.Errorf("the bucket %s: %w", delivery.ViewAnswersBucket, err) + } + _, err = kv.Put(w.ctx, key, value) + return err +} + // follow watches a bucket with its last values first and then each change, for as long as ctx lives: a bucket that is // not there yet, or a watch that ends, is tried again. func follow(ctx context.Context, js jetstream.JetStream, bucket string, model *view.Model, logger *log.Logger) { diff --git a/modules/mesh-delivery-view/internal/view/answers_test.go b/modules/mesh-delivery-view/internal/view/answers_test.go new file mode 100644 index 0000000..102d5cf --- /dev/null +++ b/modules/mesh-delivery-view/internal/view/answers_test.go @@ -0,0 +1,217 @@ +package view + +import ( + "bufio" + "encoding/json" + "fmt" + "net/http" + "net/http/httptest" + "net/url" + "strings" + "testing" + "time" + + "git.novox.be/novox/mesh-sdk/go/asks" + "git.novox.be/novox/mesh-sdk/go/delivery" +) + +// memWriter is the view's bucket answers in memory. +type memWriter struct { + puts map[string][]byte + fail error +} + +func (w *memWriter) Put(key string, value []byte) error { + if w.fail != nil { + return w.fail + } + if w.puts == nil { + w.puts = map[string][]byte{} + } + w.puts[key] = value + return nil +} + +// holdAsk is a key of the assigner's asks: the ask about a held fingerprint, as a fixture (in release 5 the assigner writes +// none yet; the value is `{"v":1,"ask":}`, as the gate's minus group and heads). +func holdAsk(t *testing.T) (Entry, asks.Ask) { + t.Helper() + key, err := delivery.HoldAskKey("plex", fpNew) + if err != nil { + t.Fatal(err) + } + a := asks.Ask{ID: delivery.AskID(key), Headline: "plex is held after a failed health check", Explanation: "ace failed the check on the new build.", Who: asks.Operator, + Expires: t0.Add(23 * time.Hour), OnExpiry: "the fingerprint stays held", + Options: []asks.Option{{ID: "release", Label: "Release", Does: "lets the move go on", Level: asks.Approve}, {ID: "keep", Label: "Keep held", Does: "keeps it held", Level: asks.Approve}}} + v, err := json.Marshal(map[string]any{"v": 1, "ask": a}) + if err != nil { + t.Fatal(err) + } + return Entry{Bucket: delivery.AssignerAsksBucket, Key: key, Value: v, At: t0}, a +} + +func TestAnAskNeedingNoProofIsOpenInTheViewWithItsOptions(t *testing.T) { // 1297/B9 + m := New() + e, a := holdAsk(t) + fill(m, e) + got := m.Snapshot().Asks + if len(got) != 1 || got[0].Key != e.Key || got[0].Headline != a.Headline || got[0].How != AnswerInTheView || strings.Join(got[0].Options, ",") != "release,keep" { + t.Fatalf("asks = %+v", got) + } + // Once the teller filled it (its answer's warrant names the ask's id) it is answered and leaves the list. + fill(m, answerToAsking(t, e.Key, a.ID)) + if got := m.Snapshot().Asks; len(got) != 0 { + t.Errorf("an answered ask is listed: %+v", got) + } +} + +func TestTheViewWritesItsAnswerUnderTheAsksViewKey(t *testing.T) { // 1297/B9a AC + m, w := New(), &memWriter{} + m.SetWriter(w) + e, _ := holdAsk(t) + fill(m, e) + if err := m.Answer(e.Key, "release", "the check was a flake"); err != nil { + t.Fatal(err) + } + raw, ok := w.puts[e.Key+".view"] + if !ok || len(w.puts) != 1 { + t.Fatalf("puts = %v, want the one key %s.view", w.puts, e.Key) + } + a, err := delivery.DecodeViewAnswer(raw) + if err != nil || a.Option != "release" || a.Words != "the check was a flake" { + t.Errorf("answer = %+v, %v", a, err) + } + // The key is the SDK's: the bucket's key parses back to the ask. + if back, err := delivery.ParseViewAnswerKey(e.Key + ".view"); err != nil || back != e.Key { + t.Errorf("parse = %q, %v", back, err) + } + // The view hears its own answer back from its bucket and shows the ask as answered, waiting for the teller. + fill(m, Entry{Bucket: delivery.ViewAnswersBucket, Key: e.Key + ".view", Value: raw, At: t0}) + got := m.Snapshot().Asks + if len(got) != 1 || got[0].Chosen != "release" { + t.Errorf("asks = %+v, want the ask with chosen release", got) + } +} + +func TestTheViewRefusesWhatItMayNotAnswer(t *testing.T) { // 1297/B9a AC + e, a := holdAsk(t) + merge := askEntry(t, "feat/x", "Merge into main: mesh-delivery#12", "Merges.") + cases := []struct { + name, key, option, want string + prep []Entry + }{ + {"the merge's Approve", merge.Key, "approve", "answer it on the phone (issue 953)", []Entry{merge}}, + {"an option the ask does not offer", e.Key, "delete", `offers no option "delete"`, []Entry{e}}, + {"an ask the view does not hold", e.Key, "release", "no open ask", nil}, + {"an ask that is no ask key", "hold.x", "release", "not an ask key", nil}, + {"an ask the teller has filled", e.Key, "release", "no open ask", []Entry{e, answerToAsking(t, e.Key, a.ID)}}, + {"no option", e.Key, " ", "no option", []Entry{e}}, + } + for _, c := range cases { + m, w := New(), &memWriter{} + m.SetWriter(w) + fill(m, c.prep...) + err := m.Answer(c.key, c.option, "") + if err == nil || !strings.Contains(err.Error(), c.want) { + t.Errorf("%s: err = %v, want it to say %q", c.name, err, c.want) + } + if len(w.puts) != 0 { + t.Errorf("%s: wrote %v", c.name, w.puts) + } + } + m := New() + e2, _ := holdAsk(t) + fill(m, e2) + if err := m.Answer(e2.Key, "release", ""); err == nil || !strings.Contains(err.Error(), "cannot write") { + t.Errorf("a view with no writer: %v", err) + } + m.SetWriter(&memWriter{fail: fmt.Errorf("the bus said no")}) + if err := m.Answer(e2.Key, "release", ""); err == nil || !strings.Contains(err.Error(), "the bus said no") { + t.Errorf("a failed write: %v", err) + } +} + +func TestTheSocketAnswersAnAsk(t *testing.T) { // 1297/B9b + path, m := served(t) + w := &memWriter{} + m.SetWriter(w) + e, _ := holdAsk(t) + fill(m, e) + if err := AnswerAsk(path, e.Key, "keep", "wait for the next build"); err != nil { + t.Fatal(err) + } + if _, ok := w.puts[e.Key+".view"]; !ok { + t.Fatalf("puts = %v", w.puts) + } + merge := askEntry(t, "feat/x", "Merge", "Merges.") + fill(m, merge) + err := AnswerAsk(path, merge.Key, "approve", "") + if err == nil || !strings.Contains(err.Error(), "answer it on the phone (issue 953)") { + t.Errorf("merge over the socket: %v", err) + } + // the old asks stay refused + c, err := dial(path) + if err != nil { + t.Fatal(err) + } + c.Write([]byte(`{"ask":"answer"}` + "\n")) + line, _ := bufio.NewReader(c).ReadString('\n') + c.Close() + if !strings.Contains(line, `"error"`) { + t.Errorf("an answer naming nothing: %q", line) + } +} + +func postAnswer(h http.Handler, form url.Values, mod func(*http.Request)) *httptest.ResponseRecorder { + req := httptest.NewRequest(http.MethodPost, "/answer", strings.NewReader(form.Encode())) + req.Host = "127.0.0.1:8471" + req.Header.Set("Content-Type", "application/x-www-form-urlencoded") + if mod != nil { + mod(req) + } + rec := httptest.NewRecorder() + h.ServeHTTP(rec, req) + return rec +} + +func TestThePageAnswersAnAskAndRefusesAForeignPost(t *testing.T) { // 1297/B9a + m, w := New(), &memWriter{} + m.SetWriter(w) + e, _ := holdAsk(t) + fill(m, e) + h := Web(m) + page := httptest.NewRecorder() + h.ServeHTTP(page, httptest.NewRequest(http.MethodGet, "/", nil)) + for _, want := range []string{`action="/answer"`, `value="release"`, `value="keep"`, e.Key} { + if !strings.Contains(page.Body.String(), want) { + t.Errorf("the page lacks %q:\n%s", want, page.Body.String()) + } + } + form := url.Values{"key": {e.Key}, "option": {"release"}, "words": {"ok"}} + // A post from another origin, or to a host name that is not the loopback's, is refused and writes nothing. + for name, mod := range map[string]func(*http.Request){ + "another origin": func(r *http.Request) { r.Header.Set("Origin", "http://evil.example") }, + "another host": func(r *http.Request) { r.Host = "evil.example" }, + } { + if rec := postAnswer(h, form, mod); rec.Code != http.StatusForbidden || len(w.puts) != 0 { + t.Errorf("%s: %d, puts %v", name, rec.Code, w.puts) + } + } + rec := postAnswer(h, form, func(r *http.Request) { r.Header.Set("Origin", "http://127.0.0.1:8471") }) + if rec.Code != http.StatusSeeOther || len(w.puts) != 1 { + t.Fatalf("answer = %d %q, puts %v", rec.Code, rec.Body.String(), w.puts) + } + // The merge's Approve is refused with its reason, and the rest of the page stays read-only. + merge := askEntry(t, "feat/x", "Merge", "Merges.") + fill(m, merge) + rec = postAnswer(h, url.Values{"key": {merge.Key}, "option": {"approve"}}, nil) + if rec.Code != http.StatusUnprocessableEntity || !strings.Contains(rec.Body.String(), "answer it on the phone (issue 953)") || len(w.puts) != 1 { + t.Errorf("merge = %d %q", rec.Code, rec.Body.String()) + } + if rec := httptest.NewRecorder(); true { + h.ServeHTTP(rec, httptest.NewRequest(http.MethodGet, "/answer", nil)) + if rec.Code != http.StatusMethodNotAllowed { + t.Errorf("GET /answer = %d", rec.Code) + } + } +} diff --git a/modules/mesh-delivery-view/internal/view/serve.go b/modules/mesh-delivery-view/internal/view/serve.go index 883fdd7..613552d 100644 --- a/modules/mesh-delivery-view/internal/view/serve.go +++ b/modules/mesh-delivery-view/internal/view/serve.go @@ -42,10 +42,14 @@ func ListenSocket(path string) (net.Listener, error) { } type ask struct { - Ask string `json:"ask"` + Ask string `json:"ask"` + Key string `json:"key,omitempty"` + Option string `json:"option,omitempty"` + Words string `json:"words,omitempty"` } -// ServeSocket answers {"ask":"rows"} on l, line-delimited JSON, until ctx ends. Any other ask is refused. +// ServeSocket answers {"ask":"rows"} on l, line-delimited JSON, until ctx ends, and {"ask":"answer","key":…,"option":…,"words":…} +// with the model's Answer (the one thing the view writes, issue 1297 B9). Any other ask is refused. func ServeSocket(ctx context.Context, l net.Listener, m *Model) error { go func() { <-ctx.Done(); l.Close() }() for { @@ -68,13 +72,46 @@ func answer(c net.Conn, m *Model) { return } var a ask - if json.Unmarshal(line, &a) != nil || a.Ask != "rows" { - json.NewEncoder(c).Encode(map[string]string{"error": `refused: this socket answers {"ask":"rows"} and nothing else`}) + if json.Unmarshal(line, &a) != nil || (a.Ask != "rows" && a.Ask != "answer") { + json.NewEncoder(c).Encode(map[string]string{"error": `refused: this socket answers {"ask":"rows"} and {"ask":"answer"} and nothing else`}) + return + } + if a.Ask == "answer" { + if err := m.Answer(a.Key, a.Option, a.Words); err != nil { + json.NewEncoder(c).Encode(map[string]string{"error": err.Error()}) + return + } + json.NewEncoder(c).Encode(map[string]string{"answered": a.Key}) return } json.NewEncoder(c).Encode(m.Snapshot()) } +// AnswerAsk asks the socket at path to write the view's answer to the open ask at key. The view's refusal is the error. +func AnswerAsk(path, key, option, words string) error { + c, err := net.DialTimeout("unix", path, 5*time.Second) + if err != nil { + return fmt.Errorf("cannot reach the view at %s: %w", path, err) + } + defer c.Close() + c.SetDeadline(time.Now().Add(10 * time.Second)) + req, _ := json.Marshal(ask{Ask: "answer", Key: key, Option: option, Words: words}) + if _, err := c.Write(append(req, '\n')); err != nil { + return err + } + var reply struct { + Answered string `json:"answered"` + Error string `json:"error"` + } + if err := json.NewDecoder(c).Decode(&reply); err != nil { + return fmt.Errorf("the view's reply is not an answer: %w", err) + } + if reply.Error != "" { + return errors.New(reply.Error) + } + return nil +} + // Ask asks the socket at path for the rows. func Ask(path string) (Snapshot, error) { c, err := net.DialTimeout("unix", path, 5*time.Second) @@ -160,7 +197,14 @@ func fingerprintText(s Snapshot) [][]string { func askText(s Snapshot) [][]string { var out [][]string for _, a := range s.Asks { - out = append(out, []string{a.Key, a.Headline, strings.Join(strings.Fields(strings.ReplaceAll(a.Words, "\n", " / ")), " "), strings.Join(a.Pulls, ", "), a.How}) + how := a.How + switch { + case a.Chosen != "": + how = "answered " + a.Chosen + " in the view; the teller fills it" + case len(a.Options) > 0: + how += ": " + strings.Join(a.Options, " or ") + } + out = append(out, []string{a.Key, a.Headline, strings.Join(strings.Fields(strings.ReplaceAll(a.Words, "\n", " / ")), " "), strings.Join(a.Pulls, ", "), how}) } return out } @@ -284,7 +328,7 @@ var page = template.Must(template.New("p").Parse(`

What each node runs

{{range .Head}}{{end}} {{range .Rows}}{{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 .Asks}}

ASKS

{{range .Asks}}{{range .Cells}}{{end}}{{end}}
ASKHEADLINEWORDSPULL REQUESTSANSWER
{{.}}{{if .Options}}
{{range .Options}} {{end}}
{{end}}
{{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}} @@ -300,6 +344,15 @@ var wantsPage = template.Must(template.New("w").Parse(`.build. - 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 . - 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 + builds map[string]buildEntry // by .build. + 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 . + asks map[string]askRecord // the gate's asks, by key + diffs map[string]diffEntry // the assigner's differences, by . + aasks map[string]assignerAskRecord // the assigner's asks (hold, restore), by key + chosen map[string]string // the options the view wrote, by ask key + writer Writer + 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 @@ -245,6 +257,11 @@ type diffEntry struct { err string } +type assignerAskRecord struct { + ask asks.Ask + err string +} + type moduleEntry struct { value delivery.Module err string @@ -269,7 +286,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{}, diffs: map[string]diffEntry{}, 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{}, aasks: map[string]assignerAskRecord{}, chosen: map[string]string{}, answers: map[string]string{}, newest: map[string]string{}, haves: map[string]haveEntry{}, nodes: map[string]nodeEntry{}, presence: map[string]presenceEntry{}, stuck: map[string]stuckEntry{}} } @@ -345,6 +362,25 @@ func (m *Model) Apply(e Entry) { } a, err := delivery.DecodeAsk(e.Value) m.asks[e.Key] = askRecord{value: a, err: errText(err)} + case delivery.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 delivery.ViewAnswersBucket: + askKey, err := delivery.ParseViewAnswerKey(e.Key) + if err != nil { + return + } + if e.Deleted { + delete(m.chosen, askKey) + return + } + if a, err := delivery.DecodeViewAnswer(e.Value); err == nil { + m.chosen[askKey] = a.Option + } case delivery.AssignerDifferencesBucket: if e.Deleted { delete(m.diffs, e.Key) @@ -521,6 +557,11 @@ func (m *Model) Snapshot() Snapshot { unreadable = append(unreadable, UnreadableRow{Bucket: delivery.AsksBucket, Key: key, Error: a.err}) } } + for key, a := range m.aasks { + if a.err != "" { + unreadable = append(unreadable, UnreadableRow{Bucket: delivery.AssignerAsksBucket, 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}) @@ -691,10 +732,96 @@ func (m *Model) askRows() []AskRow { } out = append(out, row) } + 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: a.ask.Expires, How: AnswerInTheView, Pulls: []string{}, + Chosen: m.chosen[key]} + 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 } +// decodeAssignerAsk reads a key of the assigner's asks: the router's ask under "ask" (`{"v":1,"ask":{…}}`, the gate's ask +// without its group and heads; the layout of the assigner's asks is the assigner's, issue 1297 B5). The key must be a kind of +// ask that needs no proof, and the ask must offer an option. +func decodeAssignerAsk(key string, value []byte) (asks.Ask, error) { + k, err := delivery.ParseAnyAskKey(key) + if err != nil { + return asks.Ask{}, err + } + if proof, _ := delivery.AskNeedsProof(k.Kind); proof { + return asks.Ask{}, fmt.Errorf("%s needs proof of who answered and is no ask of the assigner", key) + } + var v struct { + Ask asks.Ask `json:"ask"` + } + if err := json.Unmarshal(value, &v); err != nil { + return asks.Ask{}, fmt.Errorf("the assigner's ask %s is not readable: %w", key, err) + } + if v.Ask.ID == "" || len(v.Ask.Options) == 0 { + return asks.Ask{}, fmt.Errorf("the assigner's ask %s names no id or no option", key) + } + return v.Ask, nil +} + +// 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 +// `.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 := delivery.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.chosen[askKey], 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 asks.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 := delivery.EncodeViewAnswer(delivery.ViewAnswer{Option: option, Words: strings.TrimSpace(words)}) + if err != nil { + return err + } + return w.Put(key, body) +} + // differenceRows are the readable differences, ordered by node and module. func (m *Model) differenceRows() []DifferenceRow { out := []DifferenceRow{} diff --git a/modules/mesh-delivery-view/manifest_test.go b/modules/mesh-delivery-view/manifest_test.go index e690a08..2f25bfe 100644 --- a/modules/mesh-delivery-view/manifest_test.go +++ b/modules/mesh-delivery-view/manifest_test.go @@ -36,7 +36,13 @@ type item struct { Template string `json:"template"` } +type state struct { + Name string `json:"name"` + History int `json:"history,omitempty"` +} + type manifest struct { + State []state `json:"state"` Module string `json:"module"` Moved bool `json:"moved"` Version string `json:"version"` @@ -76,17 +82,21 @@ 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-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 + want := []string{"mesh-controller.assignments", "mesh-controller.builds", "mesh-delivery-assigner.asks", "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, B9 got := slices.Clone(m.Reads) slices.Sort(got) if !reflect.DeepEqual(got, want) { t.Errorf("reads = %v, want exactly %v", got, want) } - for _, k := range []string{"state", "emits", "consumes", "provides", "claims"} { + for _, k := range []string{"emits", "consumes", "provides", "claims"} { if _, ok := top[k]; ok { - t.Errorf("the manifest declares %q: the view writes nothing", k) + t.Errorf("the manifest declares %q: the view writes only its answers", k) } } + // The one state the view writes: the operator's answers to the asks that need no proof (ADR 0528 clause 6, 1297/B9), which the teller reads. + if want := []state{{Name: "answers", History: 5}}; !reflect.DeepEqual(m.State, want) { + t.Errorf("state = %+v, want %+v", m.State, want) + } if _, ok := m.OwnSecrets["broker"]; !ok || len(m.OwnSecrets) != 1 { t.Errorf("own-secrets = %v, want exactly broker (or no credential is issued)", m.OwnSecrets) } diff --git a/modules/mesh-delivery-view/module.json b/modules/mesh-delivery-view/module.json index c9d8718..61ed274 100644 --- a/modules/mesh-delivery-view/module.json +++ b/modules/mesh-delivery-view/module.json @@ -11,9 +11,13 @@ "builder_build", "builder_builds" ], + "state": [ + { "name": "answers", "history": 5 } + ], "reads": [ "mesh-controller.assignments", "mesh-controller.builds", + "mesh-delivery-assigner.asks", "mesh-delivery-assigner.differences", "mesh-delivery-builder.builds", "mesh-delivery-gate.asks", From e7ed48b81aaa4e41407319b091c2737578f8366e Mon Sep 17 00:00:00 2001 From: jochen Date: Fri, 23 Oct 2026 15:18:04 +0200 Subject: [PATCH 3/5] Give the command an answer verb, so the operator answers a hold or restore ask from a terminal as in the page (issue 1297 B9b) --- .../cmd/mesh-delivery-cli/main.go | 17 ++++- .../cmd/mesh-delivery-cli/main_test.go | 73 +++++++++++++++++++ 2 files changed, 87 insertions(+), 3 deletions(-) diff --git a/modules/mesh-delivery-view/cmd/mesh-delivery-cli/main.go b/modules/mesh-delivery-view/cmd/mesh-delivery-cli/main.go index fc78740..d5c1e29 100644 --- a/modules/mesh-delivery-view/cmd/mesh-delivery-cli/main.go +++ b/modules/mesh-delivery-view/cmd/mesh-delivery-cli/main.go @@ -1,5 +1,6 @@ // Command mesh-delivery-cli is the operator's command for the delivery view (novox/hq ADR 0429 §3). It asks the -// view's local socket for the rows and prints them; it holds no credential and changes nothing. +// view's local socket for the rows and prints them; it holds no credential. `answer