From e88e0c825c9886e989d948728c02dcc0fd5d767a Mon Sep 17 00:00:00 2001 From: jochen Date: Thu, 12 Nov 2026 04:56:33 +0100 Subject: [PATCH 1/5] R-5: the tool runner follows the assigner's views of the module that claims the seat, named by the seat MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The holder is read from the catalog half's entries serving a verb of mesh-module-assigner, on the branch follower's change callback; the manifest's consumes names the two views by the seat (seat:mesh-module-assigner.views.*). Builds: design 47 G §616, §428, §7 Claude-Session: https://claude.ai/code/session_01NcbgfGKaxw984XyAPfLm4h --- .../internal/console/manifest_waits_test.go | 4 +- node-tools/internal/firstview/firstview.go | 131 ++++++++++++++---- .../firstview/firstview_integration_test.go | 45 ++++-- .../internal/firstview/firstview_test.go | 89 +++++++++--- node-tools/internal/meshtest/meshtest.go | 2 +- node-tools/module.json | 4 +- 6 files changed, 208 insertions(+), 67 deletions(-) diff --git a/node-tools/internal/console/manifest_waits_test.go b/node-tools/internal/console/manifest_waits_test.go index f80ec5f..c985454 100644 --- a/node-tools/internal/console/manifest_waits_test.go +++ b/node-tools/internal/console/manifest_waits_test.go @@ -48,8 +48,8 @@ func TestTheManifestNamesTheFirstFamiliesOneByOne(t *testing.T) { // `registration-changed`, and the gate's `source-moved`; never the indexer's `views.modules` or `views.manifests`), the two // events of the seat that serves the seat list (ADR 0636 §4) and the keeper's `findings-asked`, and nothing else. wantConsumes := []string{"mesh-catalog.catalog-revision", "mesh-catalog.registration-changed", "mesh-catalog.seat-definition-changed", - "mesh-catalog.seat-definition-removed", "mesh-conditions.findings-asked", "mesh-delivery-assigner.views.assignments", - "mesh-delivery-assigner.views.seat-holders", "mesh-delivery.source-moved"} + "mesh-catalog.seat-definition-removed", "mesh-conditions.findings-asked", "seat:mesh-module-assigner.views.assignments", + "seat:mesh-module-assigner.views.seat-holders", "mesh-delivery.source-moved"} if !reflect.DeepEqual(m.Consumes, wantConsumes) { t.Errorf("the tool runner consumes %v, want %v", m.Consumes, wantConsumes) } diff --git a/node-tools/internal/firstview/firstview.go b/node-tools/internal/firstview/firstview.go index 9e60efd..706c5d7 100644 --- a/node-tools/internal/firstview/firstview.go +++ b/node-tools/internal/firstview/firstview.go @@ -20,6 +20,7 @@ import ( "sync" "time" + "git.novox.be/novox/mesh-contracts/go/seats/meshmoduleassigner" condseat "git.novox.be/novox/mesh-contracts/health-and-repair/go/seats/meshconditions" "git.novox.be/novox/mesh-delivery/contracts/go/clients/meshcatalogclient" catalogseat "git.novox.be/novox/mesh-delivery/contracts/go/seats/meshcatalog" @@ -31,27 +32,27 @@ import ( "github.com/novox/mesh-tools/node-tools/internal/bus" ) -// The owner and its first-view tools (the tools' addresses are `.`). -const Assigner = "mesh-delivery-assigner" +// AssignerSeat is the seat whose holder's views the runner follows: the holder is read from the catalog half's claims of the +// seat, never named in code (design 47 G §616, §428), and the names the console reads the two states by are the seat's +// (`.`; the manifest's `consumes` names their views families by the seat, as `seat:.views.`). +const AssignerSeat = meshmoduleassigner.Seat -// The names the console reads the assigner's two states by (`.`), as the manifest's `consumes` names their views -// families (no seat's client names a holder, novox/hq ADR 0625 §4; TestTheFirstViewsAreTheManifestsViews). const ( - AssignmentsRead = Assigner + ".assignments" - SeatHoldersRead = Assigner + ".seat-holders" + AssignmentsRead = AssignerSeat + ".assignments" + SeatHoldersRead = AssignerSeat + ".seat-holders" ) -// Spec is one first view the runner holds. -type Spec struct{ Owner, Tool, State string } +// Spec is one first view the runner holds of the seat's holder: its first-view tool and the state. +type Spec struct{ Tool, State string } -// Specs are the two owner states asked of an owner's tool; the catalog half's views are adopted by the branch follower. +// Specs are the two states asked of the holder's tool; the catalog half's views are adopted by the branch follower. var Specs = []Spec{ - {Assigner, "assignments", "assignments"}, - {Assigner, "seat-holders", "seat-holders"}, + {"assignments", "assignments"}, + {"seat-holders", "seat-holders"}, } -// Name is the name a state is read by, `.`: AssignmentsRead and SeatHoldersRead. -func (s Spec) Name() string { return s.Owner + "." + s.State } +// Name is the name a state is read by, `.`: AssignmentsRead and SeatHoldersRead. +func (s Spec) Name() string { return AssignerSeat + "." + s.State } // Views is the first views: the assigner's two, and the catalog half's per watched branch. type Views struct { @@ -62,6 +63,7 @@ type Views struct { adopted map[string]*adoption // by `.`, the views the branch follower adopted hearers map[string][]*adoption // event key `.` -> the adopted views that hear it branches *meshcatalogclient.Branches + holder *holding // the module that claims the assigner seat, once the catalog half names it ctx context.Context cancel context.CancelFunc wg sync.WaitGroup @@ -163,22 +165,6 @@ func StartWith(parent context.Context, d Deps, logf func(string, ...any)) (*View ctx, cancel := context.WithCancel(parent) v := &Views{logf: logf, ctx: ctx, cancel: cancel, followers: map[string]*views.Follower{}, byState: map[string]*views.Follower{}, adopted: map[string]*adoption{}, hearers: map[string][]*adoption{}, deps: d} - for _, s := range Specs { - s := s - call := func(tool string, body any) (json.RawMessage, error) { return d.Ask(s.Owner+"."+tool, body) } - f := &views.Follower{Owner: s.Owner, Tool: s.Tool, State: s.State, Ask: views.AskVia(call, s.Tool), Logf: logf, - Bound: d.Bound, Wait: d.Wait, - Raise: func(st views.Status) { - v.finding(unansweredFinding(v.deps.Module, v.deps.Node, st, time.Now()), st.Tool, true) - }, - Clear: func(st views.Status) { - v.finding(unansweredFinding(v.deps.Module, v.deps.Node, st, time.Now()), st.Tool, false) - }} - v.followers[s.Name()] = f - v.byState[s.State] = f - v.wg.Add(1) - go func() { defer v.wg.Done(); _ = f.Start(ctx) }() - } stopFindings := func() {} if d.StartFindings != nil { if stop, err := d.StartFindings(ctx); err != nil { @@ -194,6 +180,9 @@ func StartWith(parent context.Context, d Deps, logf func(string, ...any)) (*View logf("[mesh-tools] the catalog half is not followed (%v): the console cannot say which modules exist", err) } else { branches.Logf = func(format string, args ...any) { logf("[mesh-tools] "+format, args...) } + branches.Changed = func(_, module string, entry json.RawMessage, deleted bool, _ time.Time) { + v.claimed(module, entry, deleted) + } v.branches = branches } stop, err := d.Consume(v.hear) @@ -221,7 +210,9 @@ func (v *Views) hear(env bus.Envelope) error { if err != nil { return nil // not a views event: nothing to hold, nothing to retry } + v.mu.Lock() f := v.byState[ev.State] + v.mu.Unlock() if f == nil || f.Owner != ev.Owner { return nil } @@ -253,6 +244,13 @@ func (v *Views) finding(f condseat.Finding, tool string, raised bool) { // not answered yet, so nothing is claimed of it. func (v *Views) State(name string) (map[string][]byte, error) { v.mu.Lock() + if state, ok := strings.CutPrefix(name, AssignerSeat+"."); ok { // the seat's name stands for its holder's + if v.holder == nil { + v.mu.Unlock() + return nil, fmt.Errorf("the catalog half names no holder of the seat %s yet, so %s has no first view", AssignerSeat, name) + } + name = v.holder.module + "." + state + } f := v.followers[name] v.mu.Unlock() if f == nil { @@ -462,3 +460,78 @@ func (v *Views) Modules() (map[string]catalogseat.CatalogEntry, error) { } return out, nil } + +// holding is the views of the module that claims the assigner seat, and what stops them. +type holding struct { + module string + cancel context.CancelFunc + views []*views.Follower +} + +// claimed is the branch follower's word of one entry changed: an entry that serves a verb of the assigner seat makes its module +// the holder whose views the runner follows (a holder before it is dropped), the deletion of the holder's entry drops them +// (design 47 G §616, §428). It acts on the event alone; nothing waits on a timer for the claim. +func (v *Views) claimed(module string, entry json.RawMessage, deleted bool) { + claims := false + if !deleted { + var e catalogseat.CatalogEntry + if json.Unmarshal(entry, &e) == nil { + claims = slices.ContainsFunc(e.Serves, func(s string) bool { return strings.HasPrefix(s, AssignerSeat+".") }) + } + } + v.mu.Lock() + cur := v.holder + switch { + case claims && cur != nil && cur.module == module, !claims && (cur == nil || cur.module != module), v.ctx.Err() != nil: + v.mu.Unlock() + return + } + if cur != nil { + v.dropHolderLocked() + } + if claims { + v.startHolderLocked(module) + } + v.mu.Unlock() +} + +// startHolderLocked follows the first views of module, the holder of the seat, on the runner's own consumer. +func (v *Views) startHolderLocked(module string) { + ctx, cancel := context.WithCancel(v.ctx) + h := &holding{module: module, cancel: cancel} + d := v.deps + for _, s := range Specs { + s := s + call := func(tool string, body any) (json.RawMessage, error) { return d.Ask(module+"."+tool, body) } + f := &views.Follower{Owner: module, Tool: s.Tool, State: s.State, Ask: views.AskVia(call, s.Tool), Logf: v.logf, + Bound: d.Bound, Wait: d.Wait, + Raise: func(st views.Status) { + v.finding(unansweredFinding(d.Module, d.Node, st, time.Now()), st.Tool, true) + }, + Clear: func(st views.Status) { + v.finding(unansweredFinding(d.Module, d.Node, st, time.Now()), st.Tool, false) + }} + v.followers[module+"."+s.State] = f + v.byState[s.State] = f + h.views = append(h.views, f) + v.wg.Add(1) + go func() { defer v.wg.Done(); _ = f.Start(ctx) }() + } + v.holder = h + v.logf("[mesh-tools] the seat %s is held by %s: its views are followed", AssignerSeat, module) +} + +// dropHolderLocked stops the views of the holder and clears the findings they raised. +func (v *Views) dropHolderLocked() { + h := v.holder + v.holder = nil + h.cancel() + for _, f := range h.views { + delete(v.followers, f.Owner+"."+f.State) + delete(v.byState, f.State) + if st := f.Status(); st.Raised { + go v.finding(unansweredFinding(v.deps.Module, v.deps.Node, st, time.Now()), st.Tool, false) + } + } + v.logf("[mesh-tools] %s no longer holds the seat %s: its views are no longer followed", h.module, AssignerSeat) +} diff --git a/node-tools/internal/firstview/firstview_integration_test.go b/node-tools/internal/firstview/firstview_integration_test.go index c9fadcd..e3afd0f 100644 --- a/node-tools/internal/firstview/firstview_integration_test.go +++ b/node-tools/internal/firstview/firstview_integration_test.go @@ -17,6 +17,10 @@ import ( "git.novox.be/novox/mesh-sdk/go/views" "github.com/nats-io/nats.go" + "git.novox.be/novox/mesh-delivery/contracts/go/clients/meshcatalogclient" + catalogseat "git.novox.be/novox/mesh-delivery/contracts/go/seats/meshcatalog" + delivery "git.novox.be/novox/mesh-delivery/contracts/go/seats/meshdelivery" + "github.com/novox/mesh-tools/node-tools/internal/bus" mt "github.com/novox/mesh-tools/node-tools/internal/meshtest" ) @@ -41,7 +45,9 @@ func TestTheRunnersConsumerCarriesTheOwnersViewsFamiliesWhenTheOwnersAreDownThen if err := json.Unmarshal(raw, &manifest); err != nil { t.Fatal(err) } - streamOf := map[string]string{Assigner: catalogue.StreamBusName("assigner-views")} + // the manifest names the assigner's views by the seat; the module the catalog half says claims it is the fixture's + const holder = "assigner-fixture" + streamOf := map[string]string{"seat:" + AssignerSeat: catalogue.StreamBusName("assigner-views")} filters := map[string][]string{} families := 0 for _, c := range manifest.Consumes { @@ -53,16 +59,16 @@ func TestTheRunnersConsumerCarriesTheOwnersViewsFamiliesWhenTheOwnersAreDownThen if stream == "" { t.Fatalf("the manifest consumes %s, whose views stream the test does not know", c) } - filters[stream] = append(filters[stream], process.ModuleStreamSubject(emitter, views.EventPrefix+event)) + filters[stream] = append(filters[stream], process.ModuleStreamSubject(holder, views.EventPrefix+event)) families++ } if families != len(Specs) { t.Fatalf("the manifest consumes %d families, the runner holds %d first views", families, len(Specs)) } runner := mt.MembershipOf("node-tools", "anchor", false, nil) - for _, owner := range []string{Assigner} { - stream := streamOf[owner] - mesh.ModuleStream(t, owner, stream) + for _, seat := range []string{"seat:" + AssignerSeat} { + stream := streamOf[seat] + mesh.ModuleStream(t, holder, stream) mesh.ConsumerOn(t, stream, "anchor", "node-tools", filters[stream], 2*time.Second) runner.Hears = append(runner.Hears, bus.HeardIssued{Stream: stream, Consumer: "anchor_node-tools"}) } @@ -93,13 +99,13 @@ func TestTheRunnersConsumerCarriesTheOwnersViewsFamiliesWhenTheOwnersAreDownThen t.Cleanup(stop) // The owners are down: no tool answers, nothing is claimed, and past the bound each view is a finding. - if _, err := v.State(Assigner + ".assignments"); err == nil { + if _, err := v.State(AssignmentsRead); err == nil { t.Fatal("a state its owner has not answered is claimed") } mt.Until(t, func() error { mu.Lock() defer mu.Unlock() - if len(findings) != len(Specs)+catalogViews { + if len(findings) != catalogViews { return errFindings(len(findings)) } return nil @@ -111,10 +117,25 @@ func TestTheRunnersConsumerCarriesTheOwnersViewsFamiliesWhenTheOwnersAreDownThen t.Fatal(err) } t.Cleanup(nc.Close) + respond := func(subject string, answer any) { + if _, err := nc.Subscribe(subject, func(m *nats.Msg) { + raw, _ := json.Marshal(answer) + body, _ := json.Marshal(map[string]json.RawMessage{"result": raw}) + _ = m.Respond(body) + }); err != nil { + t.Fatal(err) + } + } + // the catalog half answers: the gate's source names main, whose entries say the fixture claims the assigner seat + respond("mesh.seat.mesh-delivery.tool.source", delivery.SourceAnswer{Branches: []string{"main"}}) + respond("mesh.seat.mesh-catalog.tool.registrations", catalogseat.RegistrationsAnswer{Registrations: []catalogseat.Registration{}}) + respond("mesh.seat.mesh-catalog.tool.revision", catalogseat.RevisionAnswer{ + Revision: &catalogseat.CatalogRevision{Revision: 3, Kind: meshcatalogclient.KindTrunk, Branch: "main"}, + Entries: []catalogseat.CatalogEntry{{Module: holder, Serves: []string{AssignerSeat + ".assignments"}}}}) for _, s := range Specs { s := s - if _, err := nc.Subscribe("mesh.mod."+s.Owner+".tool."+s.Tool, func(m *nats.Msg) { - answer, _ := json.Marshal(views.Answer{V: 1, Owner: s.Owner, State: s.State, Revision: 4, Created: time.Unix(1000, 0).UTC(), + if _, err := nc.Subscribe("mesh.mod."+holder+".tool."+s.Tool, func(m *nats.Msg) { + answer, _ := json.Marshal(views.Answer{V: 1, Owner: holder, State: s.State, Revision: 4, Created: time.Unix(1000, 0).UTC(), Values: map[string]json.RawMessage{"first": json.RawMessage(`{"n":1}`)}}) body, _ := json.Marshal(map[string]json.RawMessage{"result": answer}) _ = m.Respond(body) @@ -130,7 +151,7 @@ func TestTheRunnersConsumerCarriesTheOwnersViewsFamiliesWhenTheOwnersAreDownThen mt.Until(t, func() error { mu.Lock() defer mu.Unlock() - if len(findings) != catalogViews { // the seats' verbs are not served in this test: the catalog half stays unanswered + if len(findings) != 0 { // every view answered clears its finding return errFindings(len(findings)) } return nil @@ -138,9 +159,9 @@ func TestTheRunnersConsumerCarriesTheOwnersViewsFamiliesWhenTheOwnersAreDownThen // Each owner's next event arrives on the runner's consumer: all four families are bound. for _, s := range Specs { - ev, _ := json.Marshal(views.Event{V: 1, Owner: s.Owner, State: s.State, Key: "next", Revision: 5, Created: time.Unix(1000, 0).UTC(), + ev, _ := json.Marshal(views.Event{V: 1, Owner: holder, State: s.State, Key: "next", Revision: 5, Created: time.Unix(1000, 0).UTC(), Value: json.RawMessage(`{"n":2}`)}) - mesh.Emit(t, process.ModuleStreamSubject(s.Owner, views.EventPrefix+s.State), json.RawMessage(ev)) + mesh.Emit(t, process.ModuleStreamSubject(holder, views.EventPrefix+s.State), json.RawMessage(ev)) } for _, s := range Specs { name := s.Name() diff --git a/node-tools/internal/firstview/firstview_test.go b/node-tools/internal/firstview/firstview_test.go index 997203b..9229cb3 100644 --- a/node-tools/internal/firstview/firstview_test.go +++ b/node-tools/internal/firstview/firstview_test.go @@ -32,12 +32,14 @@ type owners struct { // catalog is the revision of the watched branch main the seat mesh-catalog answers, and modules its entries catalog int64 modules []string + // holder is the module among them that claims the assigner seat + holder string } func newOwners() *owners { return &owners{state: map[string]map[string]string{"assignments": {"anchor": `{"p":1}`}, "seat-holders": {"seat": `{"h":1}`}}, rev: map[string]uint64{"assignments": 4, "seat-holders": 2}, asked: map[string]int{}, finds: map[string]bool{}, - catalog: 3, modules: []string{"web"}} + catalog: 3, modules: []string{"web", fixtureHolder}, holder: fixtureHolder} } // Write and Clear are the runner's finding writer in a test: the findings written and not cleared, by their key. @@ -71,7 +73,11 @@ func (o *owners) ask(address string, _ any) (json.RawMessage, error) { case "seat:mesh-catalog.revision": a := catalogseat.RevisionAnswer{Revision: &catalogseat.CatalogRevision{Revision: o.catalog, Kind: meshcatalogclient.KindTrunk, Branch: "main"}} for _, m := range o.modules { - a.Entries = append(a.Entries, catalogseat.CatalogEntry{Module: m, Manifest: map[string]any{"module": m, "tools": []string{m + "_check"}}}) + e := catalogseat.CatalogEntry{Module: m, Manifest: map[string]any{"module": m, "tools": []string{m + "_check"}}} + if m == o.holder { + e.Serves = []string{AssignerSeat + ".assignments", AssignerSeat + ".seat-holders"} + } + a.Entries = append(a.Entries, e) } return json.Marshal(a) } @@ -94,14 +100,10 @@ func (o *owners) say(t *testing.T, state, key, value string, revision uint64, ga } } -func owner(state string) string { - for _, s := range Specs { - if s.State == state { - return s.Owner - } - } - return "" -} +// fixtureHolder is the module the fixture catalog half says claims the assigner seat: no name is in the code. +const fixtureHolder = "assigner-fixture" + +func owner(state string) string { return fixtureHolder } func eventually(t *testing.T, what string, ok func() bool) { t.Helper() @@ -127,37 +129,37 @@ func TestTheRunnerHoldsTheFourStatesFromTheOwnersFirstViewsAndTheirEvents(t *tes defer stop() // Down: nothing is claimed, and past the bound each unanswered view is a finding. - if _, err := v.State("mesh-delivery-assigner.assignments"); err == nil || !strings.Contains(err.Error(), "not been answered") { - t.Fatalf("a state its owner has not answered is claimed: %v", err) + if _, err := v.State(AssignmentsRead); err == nil || !strings.Contains(err.Error(), "names no holder") { + t.Fatalf("a state whose holder is not known is claimed: %v", err) } - eventually(t, "no finding past the bound", func() bool { o.mu.Lock(); defer o.mu.Unlock(); return len(o.finds) == 4 }) + eventually(t, "no finding past the bound", func() bool { o.mu.Lock(); defer o.mu.Unlock(); return len(o.finds) == 2 }) // Up: answered, the findings are cleared, and the pictures hold the owners' values. o.mu.Lock() o.down = false o.mu.Unlock() - eventually(t, "not answered", func() bool { _, err := v.State("mesh-delivery-assigner.seat-holders"); return err == nil }) + eventually(t, "not answered", func() bool { _, err := v.State(SeatHoldersRead); return err == nil }) eventually(t, "findings not cleared", func() bool { o.mu.Lock(); defer o.mu.Unlock(); return len(o.finds) == 0 }) - if got, _ := v.Entry("mesh-delivery-assigner.seat-holders", "none"); got != nil { + if got, _ := v.Entry(AssignerSeat+".seat-holders", "none"); got != nil { t.Errorf("an absent key is %q", got) } // The catalog half is followed by the shared branch follower on the runner's own consumer: the gate's source names main, // main's view is adopted, and its entries are the modules, each with its manifest (design 47 G §196, §614). - eventually(t, "the catalog half holds no modules", func() bool { m, err := v.Modules(); return err == nil && len(m) == 1 }) + eventually(t, "the catalog half holds no modules", func() bool { m, err := v.Modules(); return err == nil && len(m) == 2 }) m, _ := v.Modules() if m["web"].Manifest["module"] != "web" { t.Errorf("the module web is %+v", m["web"]) } // A catalog-revision of the branch is heard on the same consumer and asks the view once more. o.mu.Lock() - o.catalog, o.modules = 4, []string{"web", "db"} + o.catalog, o.modules = 4, []string{"web", "db", fixtureHolder} o.mu.Unlock() ev, _ := json.Marshal(catalogseat.CatalogRevisionEvent{V: 1, Branch: "main", Kind: meshcatalogclient.KindTrunk, Revision: 4}) if err := o.deliver(bus.Envelope{Key: catalogseat.Seat + "." + catalogseat.EventCatalogRevision, Body: ev}); err != nil { t.Fatal(err) } - eventually(t, "the catalog-revision was not taken", func() bool { m, err := v.Modules(); return err == nil && len(m) == 2 }) + eventually(t, "the catalog-revision was not taken", func() bool { m, err := v.Modules(); return err == nil && len(m) == 3 }) // 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] } @@ -166,7 +168,7 @@ func TestTheRunnerHoldsTheFourStatesFromTheOwnersFirstViewsAndTheirEvents(t *tes o.say(t, "assignments", "laptop", `{"p":0}`, 5, false) o.say(t, "assignments", "anchor", "", 6, false) // (a deleted mark carries no value: Deleted is the encoding's, set below) - if got, _ := v.Entry("mesh-delivery-assigner.assignments", "laptop"); string(got) != `{"p":2}` { + if got, _ := v.Entry(AssignerSeat+".assignments", "laptop"); string(got) != `{"p":2}` { t.Errorf("the next revision is not applied: %q", got) } if n := asked("assignments"); n != before { @@ -184,7 +186,7 @@ func TestTheRunnerHoldsTheFourStatesFromTheOwnersFirstViewsAndTheirEvents(t *tes assignments = s } } - if assignments.Owner != "mesh-delivery-assigner" || assignments.Tool != "assignments" || assignments.Asks < 2 || assignments.Answered.IsZero() { + if assignments.Owner != fixtureHolder || assignments.Tool != "assignments" || assignments.Asks < 2 || assignments.Answered.IsZero() { t.Errorf("the status of assignments is %+v", assignments) } manifests := asked("seat-holders") @@ -224,13 +226,19 @@ func TestTheFirstViewsAreTheManifestsViews(t *testing.T) { t.Errorf("the first view %s is read by no name here", s.Name()) } delete(want, s.Name()) - if view := s.Owner + ".views." + s.Tool; !slices.Contains(m.Consumes, view) { + // the family is named by the seat, never by its holder (design 47 G §616) + if view := "seat:" + AssignerSeat + ".views." + s.Tool; !slices.Contains(m.Consumes, view) { t.Errorf("the first view %s has no family in the manifest's consumes %v", view, m.Consumes) } } if len(want) > 0 { t.Errorf("no first view reads %v", want) } + for _, c := range m.Consumes { + if strings.Contains(c, "assigner") && !strings.HasPrefix(c, "seat:") { + t.Errorf("the manifest names the assigner by a module: %s", c) + } + } // the catalog half's follower hears these events and asks these verbs, and the manifest says so (design 47 G §614) for _, ev := range []string{"mesh-catalog.catalog-revision", "mesh-catalog.registration-changed", "mesh-delivery.source-moved"} { if !slices.Contains(m.Consumes, ev) { @@ -243,3 +251,42 @@ func TestTheFirstViewsAreTheManifestsViews(t *testing.T) { } } } + +// design 47 G §616: the holder of the assigner seat is whichever module the catalog half says serves it; when another one +// claims it the views are of the new holder, and when none does the seat's names answer that the catalog half names no holder. +func TestTheAssignersViewsFollowTheModuleThatClaimsTheSeat(t *testing.T) { + o := newOwners() + deps := Deps{Module: "node-tools", Ask: o.ask, Bound: time.Second, Wait: 10 * time.Millisecond, + Consume: func(deliver func(bus.Envelope) error) (func(), error) { o.deliver = deliver; return func() {}, nil }, Findings: o} + v, stop := StartWith(context.Background(), deps, func(string, ...any) {}) + defer stop() + eventually(t, "not answered", func() bool { _, err := v.State(AssignmentsRead); return err == nil }) + ownerOf := func() string { + for _, s := range v.Status() { + if s.State == "assignments" { + return s.Owner + } + } + return "" + } + if ownerOf() != fixtureHolder { + t.Fatalf("the views are of %q", ownerOf()) + } + revise := func(rev int64, holder string, modules ...string) { + o.mu.Lock() + o.catalog, o.modules, o.holder = rev, modules, holder + o.mu.Unlock() + ev, _ := json.Marshal(catalogseat.CatalogRevisionEvent{V: 1, Branch: "main", Kind: meshcatalogclient.KindTrunk, Revision: rev}) + if err := o.deliver(bus.Envelope{Key: catalogseat.Seat + "." + catalogseat.EventCatalogRevision, Body: ev}); err != nil { + t.Fatal(err) + } + } + revise(4, "other-holder", "web", "other-holder") + eventually(t, "the views did not move to the new holder", func() bool { return ownerOf() == "other-holder" }) + eventually(t, "not answered", func() bool { _, err := v.State(SeatHoldersRead); return err == nil }) + revise(5, "", "web") + eventually(t, "the views stayed after the claim went", func() bool { + _, err := v.State(AssignmentsRead) + return err != nil && strings.Contains(err.Error(), "names no holder") + }) +} diff --git a/node-tools/internal/meshtest/meshtest.go b/node-tools/internal/meshtest/meshtest.go index aafeef3..685b59e 100644 --- a/node-tools/internal/meshtest/meshtest.go +++ b/node-tools/internal/meshtest/meshtest.go @@ -156,7 +156,7 @@ func sdkMembership(t *testing.T, mem bus.Membership) membership.Membership { // no holder). const ( catalogHolder = "mesh-catalog" - assignerHolder = "mesh-delivery-assigner" + assignerHolder = "mesh-module-assigner" // the seat's name stands for its holder in the console's state names (firstview.AssignmentsRead) // AssignmentsBucket and SeatHoldersBucket are the assigner's states assignments and seat-holders. AssignmentsBucket = assignerHolder + "_assignments" SeatHoldersBucket = assignerHolder + "_seat-holders" diff --git a/node-tools/module.json b/node-tools/module.json index f206a4d..1c474b8 100644 --- a/node-tools/module.json +++ b/node-tools/module.json @@ -76,8 +76,8 @@ "mesh-catalog.seat-definition-changed", "mesh-catalog.seat-definition-removed", "mesh-conditions.findings-asked", - "mesh-delivery-assigner.views.assignments", - "mesh-delivery-assigner.views.seat-holders", + "seat:mesh-module-assigner.views.assignments", + "seat:mesh-module-assigner.views.seat-holders", "mesh-delivery.source-moved" ], "streams": [ From e873cb3b86145763337aed3c89bbcee9355b5671 Mon Sep 17 00:00:00 2001 From: jochen Date: Thu, 12 Nov 2026 04:57:12 +0100 Subject: [PATCH 2/5] R-7: mesh_first_views names the tool runner's own build, or says own build unknown MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Builds: design 47 G §205, §424 Claude-Session: https://claude.ai/code/session_01NcbgfGKaxw984XyAPfLm4h --- node-tools/internal/console/firstviews.go | 8 ++++++++ node-tools/internal/console/firstviews_test.go | 12 ++++++++++++ node-tools/internal/firstview/firstview.go | 10 +++++++++- 3 files changed, 29 insertions(+), 1 deletion(-) diff --git a/node-tools/internal/console/firstviews.go b/node-tools/internal/console/firstviews.go index 4a36fe1..16f44e6 100644 --- a/node-tools/internal/console/firstviews.go +++ b/node-tools/internal/console/firstviews.go @@ -6,6 +6,7 @@ import ( "log" "strings" + "git.novox.be/novox/mesh-sdk/go/process" "git.novox.be/novox/mesh-sdk/go/views" ) @@ -20,6 +21,10 @@ type FirstViewer interface { // state and since when. type entryLister interface{ Entries() any } +// ownBuilder is a FirstViewer that also says the build the tool runner itself was started with (design 47 G §205, §424): +// mesh_first_views names it, or says `own build unknown` and why. +type ownBuilder interface{ OwnBuild() process.Build } + // firstViewsAnswer is mesh_first_views and mesh_first_view_again. func (s *Surface) firstViewsAnswer(ctx context.Context, name string, args map[string]any) map[string]any { fv, ok := s.src.(FirstViewer) @@ -29,6 +34,9 @@ func (s *Surface) firstViewsAnswer(ctx context.Context, name string, args map[st } if name == verbFirstViews { out := map[string]any{"first views": fv.Status()} + if ob, ok := fv.(ownBuilder); ok { + out["own build"] = ob.OwnBuild().String() + } if el, ok := fv.(entryLister); ok { out["entries"] = el.Entries() } diff --git a/node-tools/internal/console/firstviews_test.go b/node-tools/internal/console/firstviews_test.go index abd7af9..5192c3a 100644 --- a/node-tools/internal/console/firstviews_test.go +++ b/node-tools/internal/console/firstviews_test.go @@ -7,6 +7,7 @@ import ( "strings" "testing" + "git.novox.be/novox/mesh-sdk/go/process" "git.novox.be/novox/mesh-sdk/go/views" ) @@ -14,12 +15,16 @@ import ( type viewer struct { busSource asked []string + build process.Build } func (v *viewer) Status() []views.Status { return []views.Status{{Owner: "mesh-delivery-assigner", Tool: "assignments", State: "assignments", Revision: 4, Asks: 1}} } +// OwnBuild is the build the viewer was started with. +func (v *viewer) OwnBuild() process.Build { return v.build } + // Entries are the entries by fingerprint the viewer keeps (design 47 G A-103). func (v *viewer) Entries() any { return []string{"fp:a asked"} } @@ -41,6 +46,13 @@ func TestTheConsoleShowsAndAsksAgainTheRunnersFirstViews(t *testing.T) { if got := text(s.discover(context.Background(), "mesh_first_views", nil)); !strings.Contains(got, "mesh-delivery-assigner") || !strings.Contains(got, "assignments") || !strings.Contains(got, "fp:a asked") { t.Errorf("mesh_first_views shows %s", got) } + // The tool runner's own build is named, or said unknown with the reason (design 47 G §205, §424). + for build, want := range map[string]string{"abcd1234": "abcd1234", "": "own build unknown"} { + v.build = process.ReadOwnBuild(func(string) string { return build }) + if got := text(s.discover(context.Background(), "mesh_first_views", nil)); !strings.Contains(got, `own build`) || !strings.Contains(got, want) { + t.Errorf("mesh_first_views with the build %q shows %s", build, got) + } + } got := s.discover(context.Background(), "mesh_first_view_again", map[string]any{"why": "doubted"}) if len(v.asked) != 1 || v.asked[0] != "doubted" || !strings.Contains(text(got), "asked again") { t.Errorf("mesh_first_view_again asked %v and answered %s", v.asked, text(got)) diff --git a/node-tools/internal/firstview/firstview.go b/node-tools/internal/firstview/firstview.go index 706c5d7..21f4a2f 100644 --- a/node-tools/internal/firstview/firstview.go +++ b/node-tools/internal/firstview/firstview.go @@ -27,6 +27,7 @@ import ( stdio "git.novox.be/novox/mesh-sdk/go" "git.novox.be/novox/mesh-sdk/go/keys" + "git.novox.be/novox/mesh-sdk/go/process" "git.novox.be/novox/mesh-sdk/go/views" "github.com/novox/mesh-tools/node-tools/internal/bus" @@ -86,6 +87,9 @@ type Deps struct { Findings FindingWriter // StartFindings starts that writer, and its stop ends it; nil starts none. StartFindings func(ctx context.Context) (stop func(), err error) + // Build is the build this process was started with (process.OwnBuild, read once at the start): its fingerprint, or + // why it is unknown (design 47 G §205, §424). + Build process.Build // Node is the machine the runner is on, named in its findings' subject. Node string // Bound and Wait are the follower's (zero: two minutes, five seconds). @@ -108,7 +112,7 @@ func Start(parent context.Context, conn *bus.Conn, logf func(string, ...any)) (* // state adapter, JetStream and node are this connection's, in this one place (design 47 G A-16, A-27). func DepsOf(conn *bus.Conn, logf func(string, ...any)) Deps { me := conn.Module() - d := Deps{Module: me, Node: conn.Node(), + d := Deps{Module: me, Node: conn.Node(), Build: process.OwnBuild(), Ask: func(address string, body any) (json.RawMessage, error) { got, err := conn.Ask(address, body, "") return got.Result, err @@ -287,6 +291,10 @@ func (v *Views) Status() []views.Status { return out } +// OwnBuild is the build the tool runner was started with: the fingerprint of the want its node-engine applied, or why it is +// unknown (`own build unknown: started without MESH_FINGERPRINT`). +func (v *Views) OwnBuild() process.Build { return v.deps.Build } + // Entries are the entries by fingerprint the catalog half's follower keeps, with their state and since when (design 47 G A-103). func (v *Views) Entries() any { if v.branches == nil { From 64015b59529ef61cdcf4597186d85aff4679cdaa Mon Sep 17 00:00:00 2001 From: jochen Date: Thu, 12 Nov 2026 04:57:42 +0100 Subject: [PATCH 3/5] R-8: the tool runner's own fingerprint reaches no bundle; a bundle reads the one its module is composed with MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Builds: design 47 G §519, §618 Claude-Session: https://claude.ai/code/session_01NcbgfGKaxw984XyAPfLm4h --- .../internal/runtime/own_account_test.go | 20 +++++++++++++++++++ node-tools/internal/runtime/runtime.go | 10 ++++++++-- 2 files changed, 28 insertions(+), 2 deletions(-) diff --git a/node-tools/internal/runtime/own_account_test.go b/node-tools/internal/runtime/own_account_test.go index eaf433d..a44d681 100644 --- a/node-tools/internal/runtime/own_account_test.go +++ b/node-tools/internal/runtime/own_account_test.go @@ -38,6 +38,26 @@ func TestNoBusWordReachesABundle(t *testing.T) { } } +// design 47 G §618, §519: the runtime's own fingerprint reaches no bundle; the fingerprint a module's composed words hold does. +func TestTheRuntimesOwnFingerprintReachesNoBundle(t *testing.T) { + base := []string{"MESH_FINGERPRINT=runner-build", "PATH=/usr/bin"} + for name, c := range map[string]struct { + given map[string]string + want string + }{ + "composed with none sees none": {nil, ""}, + "composed with one sees its own": {map[string]string{"MESH_FINGERPRINT": "module-build"}, "MESH_FINGERPRINT=module-build"}, + } { + env := strings.Join(BundleEnv(base, c.given, "telegram", "anchor"), "\n") + if strings.Contains(env, "runner-build") { + t.Errorf("%s: the runtime's own fingerprint reached the bundle: %s", name, env) + } + if c.want == "" && strings.Contains(env, "MESH_FINGERPRINT") || !strings.Contains(env, c.want) { + t.Errorf("%s: the environment is %s", name, env) + } + } +} + // A tool call reaches only a tool's subject: whatever key a caller names, it is never a seat's event, accept // or proof — so no tool call says a choice, a link or a code, or submits an ask, in anybody's name. func TestAToolCallNeverReachesASeatsEventAcceptOrProof(t *testing.T) { diff --git a/node-tools/internal/runtime/runtime.go b/node-tools/internal/runtime/runtime.go index f3bc178..bc982f5 100644 --- a/node-tools/internal/runtime/runtime.go +++ b/node-tools/internal/runtime/runtime.go @@ -59,6 +59,10 @@ type ListedTool struct { // NodeRuntimeModule is the module that is a machine's runtime, serving every module on it. const NodeRuntimeModule = "node-tools" +// fingerprintEnv is the word the node-engine starts each process with, the fingerprint of the want it applies (mesh-sdk's +// process.FingerprintEnv; this package imports no SDK): the runtime's own is no bundle's. +const fingerprintEnv = "MESH_FINGERPRINT" + // runtimeOnly are the words of the runtime's own environment no bundle is given. var runtimeOnly = map[string]bool{ToolEnv: true, ToolModules: true, "MESH_BROKER_FILE": true, "MESH_BROKER_URL": true, "MESH_CONSOLE_LISTEN": true, LedgersFile: true} @@ -1369,11 +1373,13 @@ func (b *moduleBus) bind(key string, stop func()) { // BundleEnv is what one module's bundle is started with: the runtime's environment without its own words, // the module's composed words over it, and the module's and machine's names. **No bus word reaches a // bundle** (security review of 2026-10-08): the credential and the bus's address are the runtime's, and a -// bundle reaches the bus only through the runtime. +// bundle reaches the bus only through the runtime. **The runtime's own fingerprint reaches none either**: a bundle reads +// only the `MESH_FINGERPRINT` its module's composed words hold, the build the node-engine applied for that module, and with +// none it has none (design 47 G §519, §618). func BundleEnv(base []string, given map[string]string, module, node string) []string { words := map[string]string{} for _, kv := range base { - if k, v, ok := strings.Cut(kv, "="); ok && !runtimeOnly[k] { + if k, v, ok := strings.Cut(kv, "="); ok && !runtimeOnly[k] && k != fingerprintEnv { words[k] = v } } From 5e373af70d2871301691f7ebd655a4f6eda09c9b Mon Sep 17 00:00:00 2001 From: jochen Date: Thu, 12 Nov 2026 05:02:04 +0100 Subject: [PATCH 4/5] R-1 rest, R-12, R-13: the capability words are mesh-sdk's, the ledger's fill and removal are answered, the artifact states the words MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit sent-get is under sent-objects-calls; mesh/ledger.fill answers the history's fill against its cap and mesh/ledger.remove removes an instance's operations, snapshot, state and index entries, both under ledger-upkeep-calls; the runtime artifact's `states` is exactly the words the runner states. Builds: design 47 G §462, §488, §545, §587, §622, §155 Claude-Session: https://claude.ai/code/session_01NcbgfGKaxw984XyAPfLm4h --- node-tools/internal/bus/ledger.go | 76 ++++++++++++++++++ node-tools/internal/ledger/ledger.go | 79 +++++++++++++++++++ node-tools/internal/ledger/ledger_test.go | 47 +++++++++++ node-tools/internal/ledger/memory.go | 53 ++++++++++++- node-tools/internal/runtime/announce_test.go | 60 +++++++++++--- node-tools/internal/runtime/busstate.go | 16 ++-- .../runtime/ledger_integration_test.go | 29 +++++++ node-tools/internal/runtime/ledgers.go | 11 +++ node-tools/internal/runtime/ledgers_test.go | 2 +- node-tools/module.json | 10 ++- 10 files changed, 364 insertions(+), 19 deletions(-) diff --git a/node-tools/internal/bus/ledger.go b/node-tools/internal/bus/ledger.go index fb38439..0d48054 100644 --- a/node-tools/internal/bus/ledger.go +++ b/node-tools/internal/bus/ledger.go @@ -10,6 +10,7 @@ import ( "strings" "time" + "git.novox.be/novox/mesh-sdk/go/buscalls" "git.novox.be/novox/mesh-sdk/go/buscompose" "git.novox.be/novox/mesh-sdk/go/membership" "github.com/nats-io/nats.go" @@ -143,6 +144,76 @@ func (h *busHistory) Read(keys string, from uint64, limit int) ([]ledger.Kept, e } } +// Purge removes every operation of one instance from the history stream, on the module's ledger connection, the one +// granted the purge of its own ledger's streams. +func (h *busHistory) Purge(key string) error { + return purgeSubject(h.read, h.write, h.stream, h.prefix+"."+key) +} + +// Fill is the history stream's fill against its cap, the cap the manifest gives the ledger. +func (h *busHistory) Fill() (buscalls.Fill, error) { + name := h.stream + if name == "" { + ctx, cancel := context.WithTimeout(context.Background(), StateTimeout) + defer cancel() + js, err := h.read.JetStream() + if err != nil { + return buscalls.Fill{}, err + } + if name, err = js.StreamNameBySubject(ctx, h.prefix+".>"); err != nil { + return buscalls.Fill{}, fmt.Errorf("no stream on the bus keeps %s: %w", h.prefix, err) + } + } + f, err := h.read.streamFill(name) + return f.Fill, err +} + +// purgeSubject purges one subject's messages from the stream that keeps it: the stream named, or found on the runner's +// own connection by its subject, purged on the ledger connection. +func purgeSubject(read, write *Conn, stream, subject string) error { + if write == nil { + return errors.New("a ledger's streams are purged only on its module's ledger connection, and none is held") + } + ctx, cancel := context.WithTimeout(context.Background(), StateTimeout) + defer cancel() + if stream == "" { + js, err := read.JetStream() + if err != nil { + return err + } + if stream, err = js.StreamNameBySubject(ctx, subject); err != nil { + return fmt.Errorf("no stream on the bus keeps %s: %w", subject, err) + } + } + // The purge is asked of the stream by its name, with no read of its info first: the ledger account is granted the + // purge of its own ledger's two streams and nothing else (PurgeAPI). + body, err := json.Marshal(map[string]string{"filter": subject}) + if err != nil { + return err + } + reply, err := write.nc.RequestWithContext(ctx, PurgeAPI(stream), body) + if err != nil { + return fmt.Errorf("purging %s from %s: %w", subject, stream, err) + } + var ack struct { + Success bool `json:"success"` + Error *struct { + Description string `json:"description"` + } `json:"error"` + } + if err := json.Unmarshal(reply.Data, &ack); err != nil || !ack.Success { + why := string(reply.Data) + if ack.Error != nil { + why = ack.Error.Description + } + return fmt.Errorf("the bus refused the purge of %s from %s: %s", subject, stream, why) + } + return nil +} + +// PurgeAPI is the bus's subject that purges a stream, the one a ledger account is granted for its own ledger's streams. +func PurgeAPI(stream string) string { return "$JS.API.STREAM.PURGE." + stream } + type busState struct { c *Conn bucket string @@ -231,6 +302,11 @@ func (s *busSnapshots) Get(key string) ([]byte, uint64, error) { return lastOn(s.read, "", s.prefix+"."+key) } +// Purge removes one instance's snapshot, on the module's ledger connection. +func (s *busSnapshots) Purge(key string) error { + return purgeSubject(s.read, s.write, "", s.prefix+"."+key) +} + func (s *busSnapshots) Put(key string, value []byte, expect uint64) error { if s.write == nil { return errors.New("a ledger's snapshot is published only on its module's ledger connection, and none is held") diff --git a/node-tools/internal/ledger/ledger.go b/node-tools/internal/ledger/ledger.go index 3088a6f..65b833d 100644 --- a/node-tools/internal/ledger/ledger.go +++ b/node-tools/internal/ledger/ledger.go @@ -58,6 +58,10 @@ type History interface { // Read is the operations of the instances keys names (a key, with `*` for one name and `>` for the rest) from a // sequence on, oldest first, at most limit of them (every one when limit is 0). Read(keys string, from uint64, limit int) ([]Kept, error) + // Purge removes every operation of the instance key names, and nothing of another (an instance's removal). + Purge(key string) error + // Fill is the history's fill against its cap, the ledger's cap (buscalls.LedgerFill; design 47 G §587). + Fill() (buscalls.Fill, error) } // State is a ledger's state bucket: each instance's record under its key. @@ -69,6 +73,8 @@ type State interface { Write(key string, value []byte, revision uint64) (uint64, error) // Keys is every key the bucket holds. Keys() ([]string, error) + // Delete removes the key; a key it does not hold is no fault. + Delete(key string) error } // Snapshots are a ledger's snapshots: the latest record of each instance, on a subject of its own, never expiring. @@ -77,6 +83,8 @@ type Snapshots interface { Get(key string) ([]byte, uint64, error) // Put writes the key's snapshot, expecting its last to be at expect (0: none); ErrRaced when it is not. Put(key string, value []byte, expect uint64) error + // Purge removes the key's snapshot; one it does not hold is no fault. + Purge(key string) error } // Ledger is one ledger of one module: its types and where it is kept. @@ -495,3 +503,74 @@ func (l *Ledger) Apply(r buscalls.ApplyRequest, author buscalls.Author) (buscall } return buscalls.Applied{Taken: true, Sequence: seq, StateBehind: behind}, nil } + +// Fill is the fill of the ledger's history against the ledger's cap (buscalls.LedgerFill; design 47 G §587): its messages, +// its bytes and the cap in bytes. The history holds every operation, and the cap is its alone: neither the fullest of the +// stores nor their sum is answered. A history that cannot say its fill is an error naming the ledger, never a fill of +// nothing. +func (l *Ledger) Fill(buscalls.LedgerRequest) (buscalls.Fill, error) { + f, err := l.History.Fill() + if err != nil { + return buscalls.Fill{}, fmt.Errorf("%s's ledger %s cannot say the fill of its history: %w", l.Module, l.Name, err) + } + return f, nil +} + +// Remove removes one instance with everything it kept (buscalls.LedgerRemove; design 47 G §587, §155): its operations, its +// snapshot, its key in the state bucket and its keys in every declared index, and answers how many operations and how many +// bytes of the history that was. An instance the ledger does not hold answers 0 and 0, and a removal that left something +// (said, and answered as an error) is made again by calling it again, which removes the rest. +func (l *Ledger) Remove(r buscalls.LedgerRemoveRequest) (buscalls.LedgerRemoved, error) { + key, err := KeyOf(r.Instance) + if err != nil { + return buscalls.LedgerRemoved{}, err + } + out := buscalls.LedgerRemoved{Instance: r.Instance} + before, err := l.History.Fill() + if err != nil { + return out, fmt.Errorf("%s's ledger %s cannot say the fill of its history, so a removal is not made: %w", l.Module, l.Name, err) + } + ops, err := l.History.Read(key, 1, 0) + if err != nil { + return out, err + } + for _, step := range []struct { + what string + remove func(string) error + }{{"operations", l.History.Purge}, {"snapshot", l.Snapshots.Purge}, {"state", l.State.Delete}, {"index entries", l.dropFromIndexes}} { + if err := step.remove(key); err != nil { + l.say("[mesh-tools] %s's ledger %s removes %s and its %s could not be removed: %v; removing it again removes what "+ + "is left", l.Module, l.Name, r.Instance, step.what, err) + return out, fmt.Errorf("%s's ledger %s removed part of %s and its %s are left: %w", l.Module, l.Name, r.Instance, step.what, err) + } + } + after, err := l.History.Fill() + if err != nil { + return out, fmt.Errorf("%s's ledger %s removed %s and cannot say the fill of its history after: %w", l.Module, l.Name, r.Instance, err) + } + out.Operations = uint64(len(ops)) + if before.Bytes > after.Bytes { + out.Bytes = before.Bytes - after.Bytes + } + return out, nil +} + +// dropFromIndexes removes the entries of the instance key names from every index bucket the ledger is bound to. +func (l *Ledger) dropFromIndexes(key string) error { + instance := InstanceOf(key) + for name, ix := range l.Indexes { + held, err := ix.Entries("", 0) + if err != nil { + return fmt.Errorf("the index %s: %w", name, err) + } + for _, e := range held { + var v buscalls.IndexEntry + if json.Unmarshal(e.Value, &v) == nil && v.Instance == instance { + if err := ix.Delete(e.Key); err != nil { + return fmt.Errorf("the index %s: %w", name, err) + } + } + } + } + return nil +} diff --git a/node-tools/internal/ledger/ledger_test.go b/node-tools/internal/ledger/ledger_test.go index 01b3017..cbc315d 100644 --- a/node-tools/internal/ledger/ledger_test.go +++ b/node-tools/internal/ledger/ledger_test.go @@ -100,3 +100,50 @@ func TestApplyAppendsDerivesAndSnapshots(t *testing.T) { }) } } + +// design 47 G §587: a ledger's fill is its history's against the ledger's cap, a store that cannot say it is an error naming +// the ledger; a removal takes an instance's operations, snapshot, state and index entries and answers what that was, an +// unknown instance answers none, and nothing of another instance goes. +func TestTheLedgerSaysItsFillAndRemovesAnInstanceWhole(t *testing.T) { + l, f := newLedger() + f.history.MaxBytes = 1 << 20 + for _, n := range []string{"a", "b"} { + applied(t, l, "decision/0290", "note", `{"says":"`+n+`"}`, owner) + } + applied(t, l, "decision/0291", "note", `{"says":"other"}`, owner) + entry, _ := json.Marshal(buscalls.IndexEntry{Instance: "decision/0290", Sequence: 2}) + other, _ := json.Marshal(buscalls.IndexEntry{Instance: "decision/0291", Sequence: 1}) + _ = f.index.Put("t.decision.0290", entry) + _ = f.index.Put("t.decision.0291", other) + + fill, err := l.Fill(buscalls.LedgerRequest{Ledger: "records"}) + if err != nil || fill.Count != 3 || fill.Bytes == 0 || fill.MaxBytes != 1<<20 { + t.Fatalf("the fill is %+v (%v), want 3 messages against 1 MiB", fill, err) + } + got, err := l.Remove(buscalls.LedgerRemoveRequest{Ledger: "records", Instance: "decision/0290"}) + if err != nil || got.Instance != "decision/0290" || got.Operations != 2 || got.Bytes == 0 { + t.Fatalf("the removal answered %+v (%v), want 2 operations and their bytes", got, err) + } + if kept, _ := f.history.Read("decision.0290", 1, 0); len(kept) != 0 { + t.Errorf("operations are left: %d", len(kept)) + } + if raw, _, _ := f.state.Get("decision.0290"); raw != nil { + t.Errorf("the state is left: %s", raw) + } + if raw, _, _ := f.snapshots.Get("decision.0290"); raw != nil { + t.Errorf("the snapshot is left: %s", raw) + } + if held, _ := f.index.Entries("", 0); len(held) != 1 || held[0].Key != "t.decision.0291" { + t.Errorf("the index holds %v, want the other instance's entry alone", held) + } + if kept, _ := f.history.Read("decision.0291", 1, 0); len(kept) != 1 { + t.Errorf("another instance lost operations: %d left", len(kept)) + } + if again, err := l.Remove(buscalls.LedgerRemoveRequest{Ledger: "records", Instance: "decision/0290"}); err != nil || again.Operations != 0 || again.Bytes != 0 { + t.Errorf("a removal said twice answered %+v (%v), want none", again, err) + } + f.history.FillFails = true + if _, err := l.Fill(buscalls.LedgerRequest{}); err == nil || !strings.Contains(err.Error(), "ledger records cannot say the fill") { + t.Errorf("a history that cannot say its fill answered %v", err) + } +} diff --git a/node-tools/internal/ledger/memory.go b/node-tools/internal/ledger/memory.go index 0b14670..1d3faf5 100644 --- a/node-tools/internal/ledger/memory.go +++ b/node-tools/internal/ledger/memory.go @@ -1,9 +1,12 @@ package ledger import ( + "errors" "sort" "strings" "sync" + + "git.novox.be/novox/mesh-sdk/go/buscalls" ) // The memory stores keep what the bus keeps, as the bus answers it, for a test of a ledger without a bus: a stream whose @@ -12,9 +15,13 @@ import ( // MemoryHistory is a history in memory. type MemoryHistory struct { - mu sync.Mutex - seq uint64 - kept []struct { + // MaxBytes is the cap its fill is said against. + MaxBytes uint64 + // FillFails makes Fill an error, as a store that cannot say its fill. + FillFails bool + mu sync.Mutex + seq uint64 + kept []struct { key string Kept } @@ -73,6 +80,32 @@ func (h *MemoryHistory) Read(keys string, from uint64, limit int) ([]Kept, error return out, nil } +func (h *MemoryHistory) Purge(key string) error { + h.mu.Lock() + defer h.mu.Unlock() + kept := h.kept[:0] + for _, k := range h.kept { + if k.key != key { + kept = append(kept, k) + } + } + h.kept = kept + return nil +} + +func (h *MemoryHistory) Fill() (buscalls.Fill, error) { + h.mu.Lock() + defer h.mu.Unlock() + if h.FillFails { + return buscalls.Fill{}, errors.New("the history does not say its fill") + } + f := buscalls.Fill{Count: uint64(len(h.kept)), MaxBytes: h.MaxBytes} + for _, k := range h.kept { + f.Bytes += uint64(len(k.Data)) + } + return f, nil +} + // Expire drops every operation of the history, as a ledger's keeping days end them. func (h *MemoryHistory) Expire() { h.mu.Lock() @@ -129,6 +162,13 @@ func (s *MemoryState) Keys() ([]string, error) { return out, nil } +func (s *MemoryState) Delete(key string) error { + s.mu.Lock() + defer s.mu.Unlock() + delete(s.kv, key) + return nil +} + // MemorySnapshots are snapshots in memory. type MemorySnapshots struct { mu sync.Mutex @@ -157,6 +197,13 @@ func (s *MemorySnapshots) Put(key string, value []byte, expect uint64) error { return nil } +func (s *MemorySnapshots) Purge(key string) error { + s.mu.Lock() + defer s.mu.Unlock() + delete(s.kv, key) + return nil +} + // MemoryIndex is an index's bucket in memory. type MemoryIndex struct { mu sync.Mutex diff --git a/node-tools/internal/runtime/announce_test.go b/node-tools/internal/runtime/announce_test.go index cc67d5b..50fea4a 100644 --- a/node-tools/internal/runtime/announce_test.go +++ b/node-tools/internal/runtime/announce_test.go @@ -3,6 +3,8 @@ package runtime import ( "encoding/json" "git.novox.be/novox/mesh-sdk/go/buscalls" + "os" + "slices" "sort" "strings" "testing" @@ -140,20 +142,60 @@ func TestTheRunnerStatesTheCapabilitiesOfItsCallsInItsPing(t *testing.T) { t.Fatalf("ping of the instance: %v", err) } var ping micro.Ping - if err := json.Unmarshal(msg.Data, &ping); err != nil || ping.Metadata[announce.Capabilities] != "bus-state-calls,ledger-calls,put-sessions,seat-settle-delay" { + if err := json.Unmarshal(msg.Data, &ping); err != nil || ping.Metadata[announce.Capabilities] != "bus-state-calls,ledger-calls,ledger-upkeep-calls,put-sessions,seat-settle-delay,sent-objects-calls" { t.Errorf("the runner states %q (%v), want its bus-state and ledger calls", ping.Metadata[announce.Capabilities], err) } } -// Design 47 G §462: the call that reads a sent object is listed under the bus state capability, and the ledger calls keep theirs. -func TestTheSentObjectCallBelongsToTheBusStateCapability(t *testing.T) { - for c, method := range map[string]string{CapBusStateCalls: buscalls.ObjectsSentGet} { - found := false - for _, m := range callFamilies[c] { - found = found || m == method +// Design 47 G §462, §488, §622: each call is stated under the capability word of its own family, the words mesh-sdk's buscalls +// names: the read of a sent object under sent-objects-calls (no longer under bus-state-calls), the ledger's fill and removal +// under ledger-upkeep-calls, the others where they were. +func TestEachCallIsStatedUnderTheWordOfItsFamily(t *testing.T) { + for method, word := range map[string]string{ + buscalls.ObjectsSentGet: buscalls.SentObjectsCalls, + buscalls.LedgerFill: buscalls.LedgerUpkeepCalls, + buscalls.LedgerRemove: buscalls.LedgerUpkeepCalls, + buscalls.LedgerApply: buscalls.LedgerCalls, + buscalls.ObjectsPut: buscalls.BusStateCalls, + } { + var in []string + for c, methods := range callFamilies { + if slices.Contains(methods, method) { + in = append(in, c) + } } - if !found { - t.Errorf("%s does not list %s", c, method) + if len(in) != 1 || in[0] != word { + t.Errorf("%s is stated under %v, want %s alone", method, in, word) } } } + +// Design 47 G §545: the words the tool runner states are exactly its manifest's artifact `states`, so the delivery plan's edge +// of §605 runs to it for each word a module's artifact `calls`. +func TestTheWordsTheRunnerStatesAreItsManifestsStates(t *testing.T) { + raw, err := os.ReadFile("../../module.json") + if err != nil { + t.Fatal(err) + } + var m struct { + Build struct { + Artifacts []struct { + Name string `json:"name"` + States []string `json:"states"` + } `json:"artifacts"` + } `json:"build"` + } + if err := json.Unmarshal(raw, &m); err != nil { + t.Fatal(err) + } + var declared []string + for _, a := range m.Build.Artifacts { + if a.Name == "runtime" { + declared = a.States + } + } + slices.Sort(declared) + if got := CallCapabilities(); !slices.Equal(got, declared) { + t.Errorf("the runner states %v, its manifest's states are %v", got, declared) + } +} diff --git a/node-tools/internal/runtime/busstate.go b/node-tools/internal/runtime/busstate.go index 5ac8bbd..d85a250 100644 --- a/node-tools/internal/runtime/busstate.go +++ b/node-tools/internal/runtime/busstate.go @@ -18,8 +18,12 @@ import ( // (announce.Capabilities) for its machine's node-engine to take into the node's record. A module whose manifest asks for one is // held off a machine whose runner does not state it. const ( - CapBusStateCalls = "bus-state-calls" - CapLedgerCalls = "ledger-calls" + CapBusStateCalls = buscalls.BusStateCalls + CapLedgerCalls = buscalls.LedgerCalls + // CapLedgerUpkeepCalls is the ledger's fill and removal (§462, §488) and CapSentObjectsCalls the read of a sent object (§488): + // each is stated once the runner answers its family, the words mesh-sdk's buscalls names (§622). + CapLedgerUpkeepCalls = buscalls.LedgerUpkeepCalls + CapSentObjectsCalls = buscalls.SentObjectsCalls // CapPutSessions is the put and get sessions' word (design 47 G §518, §622), mesh-sdk's. CapPutSessions = stdio.CapPutSessions ) @@ -27,10 +31,12 @@ const ( // callFamilies are the methods each capability stands for, the ones BusState and BusStateWatch answer. var callFamilies = map[string][]string{ CapBusStateCalls: {buscalls.ObjectsPut, buscalls.ObjectsGet, buscalls.ObjectsList, buscalls.ObjectsDelete, buscalls.ObjectsInfo, - buscalls.ObjectsWatch, buscalls.ObjectsSentGet, buscalls.StreamRead, buscalls.StreamFill, buscalls.WorkQueueSize, buscalls.WorkQueueDead}, + buscalls.ObjectsWatch, buscalls.StreamRead, buscalls.StreamFill, buscalls.WorkQueueSize, buscalls.WorkQueueDead}, CapLedgerCalls: {buscalls.LedgerApply, buscalls.LedgerGet, buscalls.LedgerHistory, buscalls.LedgerWatch, buscalls.LedgerQuery, buscalls.LedgerRebuild, buscalls.LedgerReindex, buscalls.LedgerSnapshot, buscalls.LedgerLag}, - CapPutSessions: {buscalls.ObjectsPutChunk, buscalls.ObjectsGetChunk}, + CapLedgerUpkeepCalls: {buscalls.LedgerFill, buscalls.LedgerRemove}, + CapSentObjectsCalls: {buscalls.ObjectsSentGet}, + CapPutSessions: {buscalls.ObjectsPutChunk, buscalls.ObjectsGetChunk}, } // CallCapabilities is every capability of the calls this runner answers, sorted. The delayed settle of a pulled seat @@ -111,7 +117,7 @@ func (b *moduleBus) BusState(method string, params json.RawMessage) (json.RawMes answer, err = conn.WorkQueueDead(b.module, r) } case buscalls.LedgerApply, buscalls.LedgerGet, buscalls.LedgerHistory, buscalls.LedgerQuery, buscalls.LedgerRebuild, - buscalls.LedgerReindex, buscalls.LedgerSnapshot, buscalls.LedgerLag: + buscalls.LedgerReindex, buscalls.LedgerSnapshot, buscalls.LedgerLag, buscalls.LedgerFill, buscalls.LedgerRemove: answer, err = b.ledgerCall(method, params) default: return nil, fmt.Errorf("the runtime answers no %s", method) diff --git a/node-tools/internal/runtime/ledger_integration_test.go b/node-tools/internal/runtime/ledger_integration_test.go index 802fb03..589afce 100644 --- a/node-tools/internal/runtime/ledger_integration_test.go +++ b/node-tools/internal/runtime/ledger_integration_test.go @@ -240,6 +240,35 @@ func TestTheLedgerKeepsItsPromisesOnTheBus(t *testing.T) { t.Fatalf("queried %+v, %v", q, err) } }}, + {"fill and removal are answered through the calls and remove an instance from the bus whole", "decision.gone", func(t *testing.T) { + must(t, open(t), "decision/gone", "title", `{"title":"Gone"}`) + must(t, open(t), "decision/gone", "note", `{"says":"two"}`) + must(t, open(t), "decision/stays", "note", `{"says":"kept"}`) + raw, err := m.BusState(buscalls.LedgerFill, json.RawMessage(`{"ledger":"records"}`)) + var fill buscalls.Fill + if _ = json.Unmarshal(raw, &fill); err != nil || fill.Count < 3 || fill.Bytes == 0 { + t.Fatalf("the fill answered %s (%v)", raw, err) + } + raw, err = m.BusState(buscalls.LedgerRemove, json.RawMessage(`{"ledger":"records","instance":"decision/gone"}`)) + var removed buscalls.LedgerRemoved + if _ = json.Unmarshal(raw, &removed); err != nil || removed.Operations != 2 || removed.Bytes == 0 { + t.Fatalf("the removal answered %s (%v)", raw, err) + } + l := open(t) + if last, _ := l.History.Last("decision.gone"); last != 0 { + t.Fatalf("operations are left, at %d", last) + } + if st, _, _ := l.State.Get("decision.gone"); st != nil { + t.Fatalf("the state is left: %s", st) + } + if last, _ := l.History.Last("decision.stays"); last == 0 { + t.Fatal("another instance lost its operations") + } + if raw, err = m.BusState(buscalls.LedgerRemove, json.RawMessage(`{"ledger":"records","instance":"decision/gone"}`)); err != nil || + !strings.Contains(string(raw), `"operations":0`) { + t.Fatalf("a removal said twice answered %s (%v)", raw, err) + } + }}, {"a forged author is refused", "decision.forged", func(t *testing.T) { _, err := m.BusState(buscalls.LedgerApply, json.RawMessage(`{"ledger":"records","instance":"decision/forged","operation":"title","body":{"title":"T","author":"op@shanks"},"call":"forged"}`)) if err == nil || !strings.Contains(err.Error(), "the call forged, which is not a tool call in flight") { diff --git a/node-tools/internal/runtime/ledgers.go b/node-tools/internal/runtime/ledgers.go index 114cd54..960dab5 100644 --- a/node-tools/internal/runtime/ledgers.go +++ b/node-tools/internal/runtime/ledgers.go @@ -171,6 +171,17 @@ func (b *moduleBus) ledgerCall(method string, params json.RawMessage) (any, erro var r buscalls.LedgerRequest _ = json.Unmarshal(params, &r) return l.Lag(r) + case buscalls.LedgerFill: + var r buscalls.LedgerRequest + _ = json.Unmarshal(params, &r) + return l.Fill(r) + case buscalls.LedgerRemove: + // Never on an apply: a removal is no operation of the history and has no author (design 47 G §587). + var r buscalls.LedgerRemoveRequest + if err := decode(method, params, &r); err != nil { + return nil, err + } + return l.Remove(r) } return nil, fmt.Errorf("the runtime answers no %s", method) } diff --git a/node-tools/internal/runtime/ledgers_test.go b/node-tools/internal/runtime/ledgers_test.go index 96bc96e..2fdbf66 100644 --- a/node-tools/internal/runtime/ledgers_test.go +++ b/node-tools/internal/runtime/ledgers_test.go @@ -204,7 +204,7 @@ func TestABundleToolNamedAsALedgerRepairIsRefused(t *testing.T) { // no other name; none answers "the runtime answers no ...". func TestEveryCallOfAStatedCapabilityIsAnswered(t *testing.T) { m, _, _ := aLedger(t) - if got := strings.Join(CallCapabilities(), ","); got != "bus-state-calls,ledger-calls,put-sessions,seat-settle-delay" { + if got := strings.Join(CallCapabilities(), ","); got != "bus-state-calls,ledger-calls,ledger-upkeep-calls,put-sessions,seat-settle-delay,sent-objects-calls" { t.Fatalf("the runner states %s", got) } for capability, methods := range callFamilies { diff --git a/node-tools/module.json b/node-tools/module.json index 1c474b8..076c37f 100644 --- a/node-tools/module.json +++ b/node-tools/module.json @@ -62,7 +62,15 @@ "language": "go", "system": "arch", "from": "cmd/node-tools", - "binary": "node-tools" + "binary": "node-tools", + "states": [ + "bus-state-calls", + "ledger-calls", + "ledger-upkeep-calls", + "put-sessions", + "seat-settle-delay", + "sent-objects-calls" + ] } ] }, From e63d707eac214209f434fba476c922108d98c898 Mon Sep 17 00:00:00 2001 From: jochen Date: Thu, 12 Nov 2026 05:04:18 +0100 Subject: [PATCH 5/5] R-11: mesh/publish with ack answers its sequence or {error}, mesh/subscribe with deliver new makes an ordered consumer; finding-calls is stated MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The subscription's key resolves through mesh-sdk's buscompose.EventSubjects (lead answer A-153; the sdk commit EE-S1 is cherry-picked into this change's sdk placement), a wildcard is refused. Builds: design 47 G §559, §488, §576, §622 Claude-Session: https://claude.ai/code/session_01NcbgfGKaxw984XyAPfLm4h --- node-tools/internal/bus/bus.go | 13 ++- node-tools/internal/bus/subscribenew.go | 87 +++++++++++++++++++ node-tools/internal/launch/launch.go | 56 +++++++++++- node-tools/internal/launch/sessions_test.go | 75 ++++++++++++++++ node-tools/internal/runtime/announce_test.go | 2 +- node-tools/internal/runtime/busstate.go | 5 +- node-tools/internal/runtime/ledgers_test.go | 2 +- node-tools/internal/runtime/runtime.go | 15 ++++ .../runtime/subscribenew_integration_test.go | 84 ++++++++++++++++++ node-tools/module.json | 1 + 10 files changed, 333 insertions(+), 7 deletions(-) create mode 100644 node-tools/internal/bus/subscribenew.go create mode 100644 node-tools/internal/runtime/subscribenew_integration_test.go diff --git a/node-tools/internal/bus/bus.go b/node-tools/internal/bus/bus.go index 920189e..6b5419e 100644 --- a/node-tools/internal/bus/bus.go +++ b/node-tools/internal/bus/bus.go @@ -852,6 +852,12 @@ func answerOf(data []byte) (Answered, error) { // PublishAs emits an event as a module: published into JetStream and awaited, de-duplicated by its // own id (ADR 0042). The body is the payload; the metadata rides as headers. func (c *Conn) PublishAs(module string, env Envelope) error { + _, err := c.PublishAsAcked(module, env) + return err +} + +// PublishAsAcked is PublishAs that also answers the sequence the bus acknowledged the event at. +func (c *Conn) PublishAsAcked(module string, env Envelope) (uint64, error) { // On the module's own stream when its membership names one carrying the event (ADR 0363 §2), on EVENTS // otherwise — and only there, as before, for a membership that names no stream. msg := nats.NewMsg(eventSubject(c.Membership(module), module, env.Key)) @@ -873,8 +879,11 @@ func (c *Conn) PublishAs(module string, env Envelope) error { if id := env.Headers["x-event-id"]; id != "" { opts = append(opts, nats.MsgId(id)) } - _, err := c.js.PublishMsg(msg, opts...) - return err + ack, err := c.js.PublishMsg(msg, opts...) + if err != nil { + return 0, err + } + return ack.Sequence, nil } // Say publishes one message on core NATS, kept by nobody: a heartbeat, whose loss is the next one. diff --git a/node-tools/internal/bus/subscribenew.go b/node-tools/internal/bus/subscribenew.go new file mode 100644 index 0000000..d282fa3 --- /dev/null +++ b/node-tools/internal/bus/subscribenew.go @@ -0,0 +1,87 @@ +package bus + +import ( + "context" + "encoding/json" + "fmt" + "slices" + "strings" + + "git.novox.be/novox/mesh-sdk/go/buscompose" + "github.com/nats-io/nats.go/jetstream" +) + +// SubscribeNew hears one event, named `.`, from the next message on, on an ordered consumer of its own on each +// stream that carries it, and hands every message to deliver as the envelope a module sees. The subjects the name lands on are +// mesh-sdk's one rule (buscompose.EventSubjects, design 47 G §559, A-153), so a name with a wildcard is refused, naming the rule; +// a name no stream on the bus carries is an error, never a subscription that hears nothing. stop ends every consumer. +func (c *Conn) SubscribeNew(name string, deliver func(json.RawMessage) error) (func(), error) { + subjects, err := buscompose.EventSubjects(name, "") + if err != nil { + return nil, err + } + js, err := c.JetStream() + if err != nil { + return nil, err + } + ctx, cancel := context.WithTimeout(context.Background(), StateTimeout) + defer cancel() + // The streams that carry each subject, found by the bus: a namespace no stream carries hears nothing and is skipped. + byStream := map[string][]string{} + for _, subject := range subjects { + if strings.HasSuffix(subject, ".>") { + continue + } + stream, err := js.StreamNameBySubject(ctx, subject) + if err != nil || stream == "" { + continue + } + byStream[stream] = append(byStream[stream], subject, subject+".>") + } + if len(byStream) == 0 { + return nil, fmt.Errorf("no stream on the bus carries %s (%s): nothing would be heard", name, strings.Join(subjects, ", ")) + } + var stops []func() + stopAll := func() { + for _, s := range stops { + s() + } + } + for _, stream := range slices.Sorted(mapKeys(byStream)) { + cons, err := js.OrderedConsumer(ctx, stream, jetstream.OrderedConsumerConfig{FilterSubjects: byStream[stream], + DeliverPolicy: jetstream.DeliverNewPolicy}) + if err != nil { + stopAll() + return nil, fmt.Errorf("an ordered consumer of %s on %s: %w", name, stream, err) + } + cc, err := cons.Consume(func(m jetstream.Msg) { + env, ok := envelopeOf(m.Subject(), m.Headers(), m.Data()) + if !ok { + return + } + raw, err := json.Marshal(env) + if err != nil { + return + } + if err := deliver(raw); err != nil && c.Logf != nil { + c.Logf("[mesh-tools] a subscription for new %s messages did not take one: %v", name, err) + } + }) + if err != nil { + stopAll() + return nil, fmt.Errorf("consuming %s on %s: %w", name, stream, err) + } + stops = append(stops, cc.Stop) + } + return stopAll, nil +} + +func mapKeys(m map[string][]string) func(func(string) bool) { + return func(yield func(string) bool) { + for k := range m { + if !yield(k) { + return + } + } + } +} diff --git a/node-tools/internal/launch/launch.go b/node-tools/internal/launch/launch.go index ef3826f..0dbe175 100644 --- a/node-tools/internal/launch/launch.go +++ b/node-tools/internal/launch/launch.go @@ -387,7 +387,29 @@ func Start(module, entry string, env []string, mesh Bus, logf func(string, ...an asked := time.Now() switch m.Method { case "mesh/publish": - err = mesh.Publish(m.Params) + var said struct { + Ack bool `json:"ack"` + } + _ = json.Unmarshal(m.Params, &said) + if !said.Ack { + err = mesh.Publish(m.Params) + break + } + // With ack the runtime answers only once the bus acknowledged the event: its sequence, or {error} + // (design 47 G §559). A refusal is an answer, so a writer says it as a finding's resend. + acker, ok := mesh.(interface { + PublishAcked(params json.RawMessage) (uint64, error) + }) + switch { + case !ok: + result = map[string]any{"error": "this tool runner's bus publishes no event with an acknowledgement"} + default: + if seq, perr := acker.PublishAcked(m.Params); perr != nil { + result = map[string]any{"error": perr.Error()} + } else { + result = map[string]any{"sequence": seq} + } + } case "mesh/ask": var asked json.RawMessage if asked, err = mesh.Ask(m.Params); err == nil { @@ -404,6 +426,38 @@ func Start(module, entry string, env []string, mesh Bus, logf func(string, ...an result = json.RawMessage(raw) } case "mesh/subscribe": + var said struct { + Pattern string `json:"pattern"` + Deliver string `json:"deliver"` + } + _ = json.Unmarshal(m.Params, &said) + if said.Deliver == "new" { + // An ordered consumer of its own from the next message on (design 47 G §559, A-153): each message + // is handed to this child as an event is, and the consumer ends with the child. + newer, ok := mesh.(interface { + SubscribeNew(pattern string, deliver func(json.RawMessage) error) (func(), error) + }) + if !ok { + err = errors.New("this tool runner's bus makes no subscription for new messages") + break + } + var stop func() + stop, err = newer.SubscribeNew(said.Pattern, func(envelope json.RawMessage) error { + _, err := c.ask(module, "mesh/event", map[string]any{"envelope": envelope}, EventTimeout) + return err + }) + if err == nil { + c.mu.Lock() + if c.why != nil { + c.mu.Unlock() + stop() + } else { + c.watches = append(c.watches, stop) + c.mu.Unlock() + } + } + break + } mu.Lock() subscriber = c mu.Unlock() diff --git a/node-tools/internal/launch/sessions_test.go b/node-tools/internal/launch/sessions_test.go index e4a1992..f46a5a2 100644 --- a/node-tools/internal/launch/sessions_test.go +++ b/node-tools/internal/launch/sessions_test.go @@ -2,6 +2,7 @@ package launch import ( "encoding/json" + "errors" "os" "path/filepath" "strings" @@ -66,3 +67,77 @@ func TestABundleReadsTheCapabilityWordsRunsAPutSessionAndItsChannelsClosingDrops t.Fatalf("the runner took %v", b.chunks) } } + +// findingBundle asks, as mesh-sdk's finding writer does, a publish with ack that the bus takes, one it refuses, and a +// subscription for new messages; each answer it gets is written to $FINDING_OUT, and it ends after the subscription's. +const findingBundle = `#!/bin/sh +while IFS= read -r line; do + id=$(printf '%s' "$line" | sed -n 's/.*"id":\([0-9]*\)[,}].*/\1/p') + case "$line" in + *'"method":"initialize"'*) printf '{"jsonrpc":"2.0","id":%s,"result":{}}\n' "$id" ;; + *'"method":"tools/list"'*) printf '{"jsonrpc":"2.0","id":%s,"result":{"tools":[]}}\n' "$id" + printf '{"jsonrpc":"2.0","id":"a1","method":"mesh/publish","params":{"key":"taken","ack":true,"body":{}}}\n' ;; + *'"id":"a1"'*) printf '%s\n' "$line" >>"$FINDING_OUT" + printf '{"jsonrpc":"2.0","id":"a2","method":"mesh/publish","params":{"key":"refused","ack":true,"body":{}}}\n' ;; + *'"id":"a2"'*) printf '%s\n' "$line" >>"$FINDING_OUT" + printf '{"jsonrpc":"2.0","id":"s1","method":"mesh/subscribe","params":{"pattern":"mesh-conditions.findings-asked","deliver":"new"}}\n' ;; + *'"id":"s1"'*) printf '%s\n' "$line" >>"$FINDING_OUT"; exit 0 ;; + esac +done +` + +type findingBus struct { + subscribing + mu sync.Mutex + pattern string + stopped chan struct{} +} + +func (b *findingBus) PublishAcked(params json.RawMessage) (uint64, error) { + if strings.Contains(string(params), `"refused"`) { + return 0, errors.New("the bus refused it") + } + return 42, nil +} + +func (b *findingBus) SubscribeNew(pattern string, _ func(json.RawMessage) error) (func(), error) { + b.mu.Lock() + b.pattern = pattern + b.mu.Unlock() + return func() { close(b.stopped) }, nil +} + +// Design 47 G §559: `mesh/publish` with `ack` is answered with its sequence once the bus acknowledged it, or `{error}`; +// `mesh/subscribe` with `deliver: new` makes a subscription for new messages that ends with the bundle's channel. +func TestAPublishWithAckIsAnsweredWithItsSequenceAndASubscriptionForNewMessagesEndsWithTheChannel(t *testing.T) { + dir := t.TempDir() + entry, out := filepath.Join(dir, "bundle"), filepath.Join(dir, "out") + if err := os.WriteFile(entry, []byte(findingBundle), 0o755); err != nil { + t.Fatal(err) + } + b := &findingBus{stopped: make(chan struct{})} + l, err := Start("writer", entry, append(os.Environ(), "FINDING_OUT="+out), b, func(string, ...any) {}) + if err != nil { + t.Fatal(err) + } + t.Cleanup(l.Stop) + select { + case <-b.stopped: + case <-time.After(5 * time.Second): + t.Fatal("the subscription for new messages was not ended with the bundle's channel") + } + got, _ := os.ReadFile(out) + for _, want := range []string{`"sequence":42`, `"error":"the bus refused it"`} { + if !strings.Contains(string(got), want) { + t.Errorf("the bundle was answered %s, want it to hold %s", got, want) + } + } + b.mu.Lock() + defer b.mu.Unlock() + if b.pattern != "mesh-conditions.findings-asked" { + t.Errorf("the subscription was made for %q", b.pattern) + } + if len(b.published) != 0 { + t.Errorf("a publish with ack went the unacknowledged way: %v", b.published) + } +} diff --git a/node-tools/internal/runtime/announce_test.go b/node-tools/internal/runtime/announce_test.go index 50fea4a..86356b7 100644 --- a/node-tools/internal/runtime/announce_test.go +++ b/node-tools/internal/runtime/announce_test.go @@ -142,7 +142,7 @@ func TestTheRunnerStatesTheCapabilitiesOfItsCallsInItsPing(t *testing.T) { t.Fatalf("ping of the instance: %v", err) } var ping micro.Ping - if err := json.Unmarshal(msg.Data, &ping); err != nil || ping.Metadata[announce.Capabilities] != "bus-state-calls,ledger-calls,ledger-upkeep-calls,put-sessions,seat-settle-delay,sent-objects-calls" { + if err := json.Unmarshal(msg.Data, &ping); err != nil || ping.Metadata[announce.Capabilities] != "bus-state-calls,finding-calls,ledger-calls,ledger-upkeep-calls,put-sessions,seat-settle-delay,sent-objects-calls" { t.Errorf("the runner states %q (%v), want its bus-state and ledger calls", ping.Metadata[announce.Capabilities], err) } } diff --git a/node-tools/internal/runtime/busstate.go b/node-tools/internal/runtime/busstate.go index d85a250..ccf7ed7 100644 --- a/node-tools/internal/runtime/busstate.go +++ b/node-tools/internal/runtime/busstate.go @@ -40,9 +40,10 @@ var callFamilies = map[string][]string{ } // CallCapabilities is every capability of the calls this runner answers, sorted. The delayed settle of a pulled seat -// message names a fate, not a method, so it stands beside the method families (novox/hq ADR 0664 §5). +// message names a fate, not a method, so it stands beside the method families (novox/hq ADR 0664 §5); so does `finding-calls`, +// the acknowledged `mesh/publish` and the `mesh/subscribe` for new messages the launch answers (design 47 G §559). func CallCapabilities() []string { - out := []string{stdio.CapSeatSettleDelay} + out := []string{stdio.CapSeatSettleDelay, stdio.CapFindingCalls} for c := range callFamilies { out = append(out, c) } diff --git a/node-tools/internal/runtime/ledgers_test.go b/node-tools/internal/runtime/ledgers_test.go index 2fdbf66..324d054 100644 --- a/node-tools/internal/runtime/ledgers_test.go +++ b/node-tools/internal/runtime/ledgers_test.go @@ -204,7 +204,7 @@ func TestABundleToolNamedAsALedgerRepairIsRefused(t *testing.T) { // no other name; none answers "the runtime answers no ...". func TestEveryCallOfAStatedCapabilityIsAnswered(t *testing.T) { m, _, _ := aLedger(t) - if got := strings.Join(CallCapabilities(), ","); got != "bus-state-calls,ledger-calls,ledger-upkeep-calls,put-sessions,seat-settle-delay,sent-objects-calls" { + if got := strings.Join(CallCapabilities(), ","); got != "bus-state-calls,finding-calls,ledger-calls,ledger-upkeep-calls,put-sessions,seat-settle-delay,sent-objects-calls" { t.Fatalf("the runner states %s", got) } for capability, methods := range callFamilies { diff --git a/node-tools/internal/runtime/runtime.go b/node-tools/internal/runtime/runtime.go index bc982f5..c95d14e 100644 --- a/node-tools/internal/runtime/runtime.go +++ b/node-tools/internal/runtime/runtime.go @@ -876,6 +876,21 @@ func (b *moduleBus) Publish(params json.RawMessage) error { return b.all.conn.PublishAs(b.module, env) } +// PublishAcked is Publish that answers the sequence the bus acknowledged the event at (`mesh/publish` with `ack`, design 47 G §559). +func (b *moduleBus) PublishAcked(params json.RawMessage) (uint64, error) { + var env bus.Envelope + if err := json.Unmarshal(params, &env); err != nil { + return 0, fmt.Errorf("not an event envelope: %w", err) + } + return b.all.conn.PublishAsAcked(b.module, env) +} + +// SubscribeNew hands the bundle one event's messages published from now on, on an ordered consumer of its own (`mesh/subscribe` +// with `deliver: new`, design 47 G §559); the consumer ends with the bundle's channel. +func (b *moduleBus) SubscribeNew(pattern string, deliver func(json.RawMessage) error) (func(), error) { + return b.all.conn.SubscribeNew(pattern, deliver) +} + // Ask calls a tool as the module: `{key, body}`, answered with the tool's result (ADR 0198). func (b *moduleBus) Ask(params json.RawMessage) (json.RawMessage, error) { var asked struct { diff --git a/node-tools/internal/runtime/subscribenew_integration_test.go b/node-tools/internal/runtime/subscribenew_integration_test.go new file mode 100644 index 0000000..9dc1d14 --- /dev/null +++ b/node-tools/internal/runtime/subscribenew_integration_test.go @@ -0,0 +1,84 @@ +//go:build integration + +package runtime + +import ( + "context" + "encoding/json" + "strings" + "sync" + "testing" + + "github.com/nats-io/nats.go" + "github.com/nats-io/nats.go/jetstream" + + "github.com/novox/mesh-tools/node-tools/internal/bus" + mt "github.com/novox/mesh-tools/node-tools/internal/meshtest" +) + +// Design 47 G §559, A-153: against the real bus a publish with ack answers the sequence the bus gave it, and a subscription +// for new messages hears only what is published after it was made, on the subject mesh-sdk's one rule names for the event; +// a wildcard is refused. +func TestAnAckedPublishAnswersItsSequenceAndANewSubscriptionHearsOnlyWhatComesAfter(t *testing.T) { + mt.New(t) + nc, err := nats.Connect(mt.URL(t)) + if err != nil { + t.Fatal(err) + } + t.Cleanup(nc.Close) + js, _ := jetstream.New(nc) + ctx := context.Background() + if _, err := js.CreateStream(ctx, jetstream.StreamConfig{Name: "SEATEVENTS", Subjects: []string{"mesh.seat.mesh-conditions.event.>"}}); err != nil { + t.Fatal(err) + } + conn := connect(t, "node-tools", "anchor") + publish := func(body string) { + t.Helper() + if _, err := js.Publish(ctx, "mesh.seat.mesh-conditions.event.findings-asked", []byte(body)); err != nil { + t.Fatal(err) + } + } + publish(`{"before":true}`) + + var mu sync.Mutex + var heard []string + stop, err := conn.SubscribeNew("mesh-conditions.findings-asked", func(raw json.RawMessage) error { + var env bus.Envelope + _ = json.Unmarshal(raw, &env) + mu.Lock() + heard = append(heard, env.Key+" "+string(env.Body)) + mu.Unlock() + return nil + }) + if err != nil { + t.Fatal(err) + } + defer stop() + publish(`{"after":true}`) + publish(`{"after":2}`) + mt.Until(t, func() error { + mu.Lock() + defer mu.Unlock() + if len(heard) < 2 { + return context.DeadlineExceeded + } + return nil + }) + mu.Lock() + if len(heard) != 2 || heard[0] != `mesh-conditions.findings-asked {"after":true}` { + t.Errorf("heard %v, want the messages after the subscription alone, in order", heard) + } + mu.Unlock() + + for name, want := range map[string]string{"mesh-conditions.*": "[A-Za-z0-9_-]"} { + if _, err := conn.SubscribeNew(name, func(json.RawMessage) error { return nil }); err == nil || !strings.Contains(err.Error(), want) { + t.Errorf("a subscription for %s answered %v, want one naming %q", name, err, want) + } + } + + first, err := conn.PublishAsAcked("node-tools", bus.Envelope{Key: "finding-written", Body: json.RawMessage(`{}`)}) + second, err2 := conn.PublishAsAcked("node-tools", bus.Envelope{Key: "finding-written", Body: json.RawMessage(`{}`)}) + if err != nil || err2 != nil || first == 0 || second != first+1 { + t.Errorf("acknowledged publishes answered %d (%v) and %d (%v), want consecutive sequences", first, err, second, err2) + } +} diff --git a/node-tools/module.json b/node-tools/module.json index 076c37f..fbc4950 100644 --- a/node-tools/module.json +++ b/node-tools/module.json @@ -65,6 +65,7 @@ "binary": "node-tools", "states": [ "bus-state-calls", + "finding-calls", "ledger-calls", "ledger-upkeep-calls", "put-sessions",