From 0b0f879e86f1225e7d0a8428989fdc5f31a90308 Mon Sep 17 00:00:00 2001 From: jochen Date: Thu, 12 Nov 2026 05:11:52 +0100 Subject: [PATCH] E-4: a misspelt event name is refused, not waited on for ever (issue 1966) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Before it subscribes, mesh_wait checks each name's emitter against the catalog half the tool runner holds (its seats' definitions and its modules' entries), refusing one neither knows and naming the half and its revisions; a known emitter's undeclared event is waited on and its timed-out answer says so; while the half is not yet read the answer says the names were unchecked. The first views answer the halves by branch for it. Builds: design 34 every-event §14, §19 (rows of the table of refusals) Claude-Session: https://claude.ai/code/session_01NcbgfGKaxw984XyAPfLm4h --- node-tools/internal/console/wait.go | 110 +++++++++++++++++- node-tools/internal/console/wait_test.go | 35 +++++- node-tools/internal/firstview/firstview.go | 19 +++ .../internal/firstview/firstview_test.go | 4 + 4 files changed, 166 insertions(+), 2 deletions(-) diff --git a/node-tools/internal/console/wait.go b/node-tools/internal/console/wait.go index c35af73..a93a16d 100644 --- a/node-tools/internal/console/wait.go +++ b/node-tools/internal/console/wait.go @@ -17,7 +17,9 @@ import ( "strings" "time" + "git.novox.be/novox/mesh-delivery/contracts/go/clients/meshcatalogclient" "git.novox.be/novox/mesh-sdk/go/buscompose" + "git.novox.be/novox/mesh-sdk/go/catalogue" ) // WaitDefault is how long a wait lasts when no timeout is given, and WaitLongest the most it may last @@ -75,12 +77,29 @@ func (s *Surface) waitAnswer(ctx context.Context, args map[string]any) map[strin return failure(fmt.Sprintf("mesh_wait names at most %d events in one call, not %d: each name is six "+ "subscriptions for the call (design 34 every-event §13)", waitMostNames, len(names))) } - var subjects []string + known, unchecked := s.emitters() + var subjects, undeclared []string for _, name := range names { subs, err := buscompose.EventSubjects(name, from) if err != nil { return failure("mesh_wait may not wait on it: " + err.Error()) } + // A misspelt emitter is refused, never waited on for ever; a known emitter's undeclared event is waited on and + // its timed-out answer says so (design 34 every-event §14). + if unchecked == "" { + emitter, event, _ := strings.Cut(name, ".") + seatEvents, isSeat := known.seats[emitter] + moduleEvents, isModule := known.modules[emitter] + isSeat = isSeat && from != "module" && from != "stream" + isModule = isModule && from != "seat" + if !isSeat && !isModule { + return failure(fmt.Sprintf("mesh_wait may not wait on %q: %s is neither a seat nor a module of %s", + name, emitter, known.source)) + } + if !(isSeat && declares(seatEvents, event)) && !(isModule && declares(moduleEvents, event)) { + undeclared = append(undeclared, name) + } + } for _, subject := range subs { if !slices.Contains(subjects, subject) { subjects = append(subjects, subject) @@ -96,6 +115,10 @@ func (s *Surface) waitAnswer(ctx context.Context, args map[string]any) map[strin if !ok { answer = map[string]any{"timed-out": true, "waited": timeout.String(), "subjects": names, "read": "nothing matching was heard; read the state once from the verb that owns it, then wait again"} + if len(undeclared) > 0 { + answer["undeclared"] = fmt.Sprintf("no declaration names %s: no seat definition and no module's entry in %s "+ + "declares that event", strings.Join(undeclared, ", "), known.source) + } } else { name, from, _ := buscompose.EventName(heard.Subject) h := map[string]any{"subject": name, "from": from, "body": heard.Body, "at": heard.At.Format(time.RFC3339Nano)} @@ -109,6 +132,9 @@ func (s *Surface) waitAnswer(ctx context.Context, args map[string]any) map[strin } answer = map[string]any{"heard": h, "read": read} } + if unchecked != "" { + answer["unchecked"] = "the names were not checked: " + unchecked + } // A subscription the bus refused while others stood is said (design 34 every-event §16). if len(refused) > 0 { answer["refused"] = refused @@ -116,6 +142,88 @@ func (s *Surface) waitAnswer(ctx context.Context, args map[string]any) map[strin return answerText(answer) } +// halfReader is a Source that also holds the catalog half of each followed branch (firstview.Views.Halves): mesh_wait +// checks an event name's emitter against it (design 34 every-event §14). +type halfReader interface { + Halves() (map[string]meshcatalogclient.Half, error) +} + +// knownEmitters is what the catalog half says emits what: each seat's declared events and each module's, and the +// source a refusal names, with its revisions. +type knownEmitters struct { + seats, modules map[string][]string + source string +} + +// emitters is the catalog half's emitters, or why they are not checked: the half not yet read (its first view, +// mesh_first_views), or no half held at all. +func (s *Surface) emitters() (known knownEmitters, unchecked string) { + hr, ok := s.src.(halfReader) + if !ok { + return known, "this tool runner holds no catalog half" + } + halves, err := hr.Halves() + if err != nil { + return known, "the catalog half is not yet read (" + err.Error() + "); its first view is in mesh_first_views" + } + seatEvents := map[string][]string{} // a seat's events, as its definition in the seat list declares them + for _, st := range catalogue.Seats() { + seatEvents[st.Name] = st.Emits + } + known.seats, known.modules = map[string][]string{}, map[string][]string{} + branches := make([]string, 0, len(halves)) + for b := range halves { + branches = append(branches, b) + } + slices.Sort(branches) + var revisions []string + for _, b := range branches { + half := halves[b] + revisions = append(revisions, fmt.Sprintf("revision %d of %s", half.Revision.Revision, b)) + for _, d := range half.Revision.Definitions { + known.seats[d.Seat] = seatEvents[d.Seat] + } + for m, entry := range half.Entries { + if _, taken := known.modules[m]; !taken { + known.modules[m] = eventsDeclared(entry.Manifest) + } + } + } + known.source = "the catalog half (mesh-catalog.revision, " + strings.Join(revisions, "; ") + ")" + return known, "" +} + +// eventsDeclared is the events a module's manifest declares: its `emits` and its streams' `events`. +func eventsDeclared(manifest map[string]any) []string { + var out []string + add := func(list any) { + items, _ := list.([]any) + for _, x := range items { + if e, ok := x.(string); ok { + out = append(out, e) + } + } + } + add(manifest["emits"]) + streams, _ := manifest["streams"].([]any) + for _, st := range streams { + if m, ok := st.(map[string]any); ok { + add(m["events"]) + } + } + return out +} + +// declares says whether one of the declared events is the event a name gives, or one its name is followed by. +func declares(declared []string, event string) bool { + for _, d := range declared { + if buscompose.Overlaps(d, event) || buscompose.Overlaps(d, event+".>") { + return true + } + } + return false +} + // waitNames is the event names a call gives: a list of names, or one name. func waitNames(given any) ([]string, error) { var names []string diff --git a/node-tools/internal/console/wait_test.go b/node-tools/internal/console/wait_test.go index a19522e..21bd105 100644 --- a/node-tools/internal/console/wait_test.go +++ b/node-tools/internal/console/wait_test.go @@ -12,6 +12,8 @@ import ( "testing" "time" + "git.novox.be/novox/mesh-delivery/contracts/go/clients/meshcatalogclient" + catalogseat "git.novox.be/novox/mesh-delivery/contracts/go/seats/meshcatalog" "github.com/nats-io/nats-server/v2/server" "github.com/nats-io/nats.go" @@ -35,6 +37,14 @@ func waitingRuntime(t *testing.T) (*Surface, *mt.Mesh) { return NewSurface(conn, busSource{conn}, "desk.node-tools"), mesh } +// halfSource is a Source that also holds a catalog half, as the tool runner's first views do. +type halfSource struct { + Source + halves map[string]meshcatalogclient.Half +} + +func (h halfSource) Halves() (map[string]meshcatalogclient.Half, error) { return h.halves, nil } + // listening waits until the surface's connection holds n more subscriptions than before: a wait has subscribed. func listening(t *testing.T, s *Surface, before, n int) { t.Helper() @@ -157,7 +167,8 @@ func TestMeshWaitSaysItTimedOutAndKeepsNothing(t *testing.T) { if took := time.Since(start); took < time.Second || took > 3*time.Second { t.Errorf("a one-second wait took %s", took) } - if isErr || !strings.Contains(text, `"timed-out": true`) { + // Before the catalog half is read, names are not checked, and the answer says so (design 34 every-event §14). + if isErr || !strings.Contains(text, `"timed-out": true`) || !strings.Contains(text, `"unchecked"`) { t.Errorf("a wait that heard nothing: %s", text) } if after := s.conn.Subscriptions(); after != before { @@ -211,6 +222,28 @@ func TestMeshWaitRefusesLoudlyAndNeverLoops(t *testing.T) { } } + // E §14: against the catalog half, a misspelt emitter is refused naming the half and its revision, a known emitter's + // undeclared event is waited on and its timeout says so, and a declared one says nothing of it. + half := NewSurface(s.conn, halfSource{busSource{s.conn}, map[string]meshcatalogclient.Half{"main": { + Revision: catalogseat.CatalogRevision{Revision: 7, Definitions: []catalogseat.CatalogRevisionDefinitionsItem{{Seat: "mesh-catalog"}}}, + Entries: map[string]catalogseat.CatalogEntry{"mesh-delivery-view": {Module: "mesh-delivery-view", + Manifest: map[string]any{"emits": []any{"delivered"}}}}, + }}}, "desk.node-tools") + for _, c := range []struct { + name, says string + refused bool + }{ + {"mesh-delivry-view.delivered", "mesh-delivry-view is neither a seat nor a module of the catalog half (mesh-catalog.revision, revision 7 of main)", true}, + {"mesh-delivery-view.planned", "no declaration names mesh-delivery-view.planned", false}, + {"mesh-catalog.catalog-revision", "", false}, + } { + text, isErr := callWait(half, map[string]any{"subjects": []any{c.name}, "timeout": 1}) + if isErr != c.refused || c.says != "" && !strings.Contains(text, c.says) || + c.says == "" && (strings.Contains(text, "undeclared") || strings.Contains(text, "unchecked")) { + t.Errorf("%s against the catalog half: %s", c.name, text) + } + } + // The bus refusing every subscription of the call: answered after the timeout, in §16's words. b, err := server.NewServer(&server.Options{Host: "127.0.0.1", Port: server.RANDOM_PORT, NoLog: true, NoSigs: true, Users: []*server.User{{Username: "desk.node-tools", Password: "pw", Permissions: &server.Permissions{ diff --git a/node-tools/internal/firstview/firstview.go b/node-tools/internal/firstview/firstview.go index 9e60efd..ff5b5eb 100644 --- a/node-tools/internal/firstview/firstview.go +++ b/node-tools/internal/firstview/firstview.go @@ -437,6 +437,25 @@ func (w wiring) Drop(owner, state string) error { return nil } +// Halves is the catalog half of each followed branch, by branch: its revision, with the seat definitions it was judged +// with, and its entries; mesh_wait checks an event name's emitter against them (design 34 every-event §14). An error says +// the half has no first view yet, never that no module exists. +func (v *Views) Halves() (map[string]meshcatalogclient.Half, error) { + if v.branches == nil { + return nil, fmt.Errorf("the tool runner does not follow the catalog half") + } + if err := v.branches.Ready(); err != nil { + return nil, err + } + out := map[string]meshcatalogclient.Half{} + for _, branch := range v.branches.Followed() { + if c, ok := v.branches.Catalog(branch); ok { + out[branch], _ = c.View() // a branch at none-yet is ready and empty (§614) + } + } + return out, nil +} + // Modules is every module the catalog half holds, by name, with its entry: the entries of the watched branches' newest trunk // revisions, the branch followed first naming a module more than one hold (design 47 G §196, §594). An error says the half has // no first view yet, never that no module exists. diff --git a/node-tools/internal/firstview/firstview_test.go b/node-tools/internal/firstview/firstview_test.go index 997203b..18c7698 100644 --- a/node-tools/internal/firstview/firstview_test.go +++ b/node-tools/internal/firstview/firstview_test.go @@ -158,6 +158,10 @@ func TestTheRunnerHoldsTheFourStatesFromTheOwnersFirstViewsAndTheirEvents(t *tes t.Fatal(err) } eventually(t, "the catalog-revision was not taken", func() bool { m, err := v.Modules(); return err == nil && len(m) == 2 }) + // And the halves by branch, with their revision, which mesh_wait names checking an emitter (design 34 every-event §14). + if h, err := v.Halves(); err != nil || h["main"].Revision.Revision != 4 || len(h["main"].Entries) != 2 { + t.Errorf("the halves: %+v, %v", h, err) + } // Events: the next revision is applied; an old one passed over; a deletion removes; none of it asks again. asked := func(tool string) int { o.mu.Lock(); defer o.mu.Unlock(); return o.asked[tool] }