From a196bf63230771db315084026802c8c371547dcc Mon Sep 17 00:00:00 2001 From: jochen Date: Thu, 12 Nov 2026 11:15:02 +0100 Subject: [PATCH] Area 6 (join 11): every AnswerSeat caller follows a seat list and its grant AnswerSeat checks a call's caller against the seat's definition its caller reads from the seat list it follows (A-186). The bus module's keeper hands it the mesh-broker definition its seat list holds once heard; mesh-recovery, which followed none, follows one over its connection and an event hub, read again once its events are live, and its manifest and its process invoke seat-definitions and consume the list's two events. Its whole test answers the seat list. Builds: design 47 G lead answers A-179, A-186 (join 11 area 6) Claude-Session: https://claude.ai/code/session_01NcbgfGKaxw984XyAPfLm4h --- .../mesh-recovery/cmd/mesh-recovery/main.go | 25 +++++++++- .../cmd/mesh-recovery/main_test.go | 42 +++++++++++++++++ .../mesh-recovery/whole_integration_test.go | 46 +++++++++++++++++-- .../internal/recovery/manifest_test.go | 12 ++++- modules/mesh-recovery/module.json | 6 ++- modules/nats/cmd/nats-tools/keep.go | 11 ++++- modules/nats/cmd/nats-tools/keep_test.go | 46 +++++++++++++++++++ .../cmd/nats-tools/keeper_integration_test.go | 12 ++++- 8 files changed, 190 insertions(+), 10 deletions(-) diff --git a/modules/mesh-recovery/cmd/mesh-recovery/main.go b/modules/mesh-recovery/cmd/mesh-recovery/main.go index 9e9a1a69e..d85dc1369 100644 --- a/modules/mesh-recovery/cmd/mesh-recovery/main.go +++ b/modules/mesh-recovery/cmd/mesh-recovery/main.go @@ -20,6 +20,7 @@ import ( "git.novox.be/novox/mesh-contracts/health-and-repair/go/clients/meshhandactsclient" "git.novox.be/novox/mesh-contracts/health-and-repair/go/seats/meshconditions" "git.novox.be/novox/mesh-sdk/go/process" + "git.novox.be/novox/mesh-sdk/go/seatdef" "git.novox.be/novox/mesh-sdk/go/seatlist" "git.novox.be/novox/mesh-sdk/go/views" @@ -168,7 +169,16 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro for _, x := range e.Start(ctx) { logger.Printf("the aside %s of %s's %s holds data the live bus lacks: recovery_restore_aside puts it back", x.Aside.ID, x.Module, x.Kept) } - stopAnswers, err := conn.AnswerSeat(recovery.Seat, e.Handlers()) + // The seat list, read over the connection and heard on the module's event hub (seatlist.Part), which the seat's verbs check + // their callers against (A-179, A-186): registered before the hub starts, read again once its events are live. + hub := process.NewEventHub(logger.Printf) + seats := seatlist.NewPart("recovery", nil) + seats.Follow(seatlist.ProcessBus(conn.Call, hub), nil, logger.Printf) + if err := hub.StartUntilBound(ctx, js, account, nil, seats.EventsLive); err != nil { + logger.Printf("the consumer %s is not bound yet (%v): bound once the bus makes it, the seat list read then", account.EventConsumerName(), err) + } + defer hub.Stop() + stopAnswers, err := conn.AnswerSeat(recovery.Seat, seatDefinition(seats, recovery.Seat), e.Handlers()) if err != nil { return fmt.Errorf("the seat %s's verbs cannot be answered: %w", recovery.Seat, err) } @@ -193,3 +203,16 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro stopAnswers() return nil } + +// seatDefinition is the definition of seat the seat list holds (A-186): false until the seat mesh-catalog has answered the +// list, so the module answers a verb only once its seat list is read, and false for a seat it does not hold. +func seatDefinition(seats *seatlist.Part, seat string) func() (seatdef.Definition, bool) { + return func() (seatdef.Definition, bool) { + r := seats.Reader() + if r == nil || !r.Status().Heard { + return seatdef.Definition{}, false + } + d, ok := r.Definitions()[seat] + return d, ok + } +} diff --git a/modules/mesh-recovery/cmd/mesh-recovery/main_test.go b/modules/mesh-recovery/cmd/mesh-recovery/main_test.go index e0e4482d8..5205d8c1c 100644 --- a/modules/mesh-recovery/cmd/mesh-recovery/main_test.go +++ b/modules/mesh-recovery/cmd/mesh-recovery/main_test.go @@ -2,6 +2,8 @@ package main import ( "context" + "encoding/json" + "errors" "os" "path/filepath" "strings" @@ -11,13 +13,53 @@ import ( "git.novox.be/novox/mesh-contracts/go/seats/meshbroker" "git.novox.be/novox/mesh-contracts/go/seats/nodeengine" stdio "git.novox.be/novox/mesh-sdk/go" + "git.novox.be/novox/mesh-sdk/go/busretry" "git.novox.be/novox/mesh-sdk/go/catalogue" "git.novox.be/novox/mesh-sdk/go/process" + "git.novox.be/novox/mesh-sdk/go/seatlist" ) +// listingCatalog is the seat mesh-catalog as the reader asks it: an empty listing, so the bootstrapped definitions stand, or, +// silent, no answer at all. +type listingCatalog struct{ silent bool } + +func (c listingCatalog) Ask(string, any) (json.RawMessage, error) { + if c.silent { + return nil, errors.New("no responders") + } + return json.Marshal(seatlist.Listing{Revision: 1}) +} + +func (listingCatalog) Subscribe(string, func(string, json.RawMessage) error) error { return nil } + // The process's own readings of the mesh (design 47 G §554, §563, A-121, A-132): the bus names of a kept data from its // manifest, the fingerprint a module runs at from the node-engine seat's running, and whether a dump of the bus module runs. func TestWhatTheProcessReads(t *testing.T) { + // A-179, A-186: the definition the seat's verbs check their callers against is the seat list's, none before the reader + // starts or while the seat mesh-catalog has not answered, and none for a seat the list does not hold. + t.Run("the seat's definition from the seat list once read (A-186)", func(t *testing.T) { + t.Cleanup(busretry.Waits(20*time.Millisecond, time.Millisecond, 5*time.Millisecond)) + follow := func(silent bool) *seatlist.Part { + p := seatlist.NewPart("recovery", nil) + if r := p.Follow(listingCatalog{silent: silent}, nil, t.Logf); r != nil { + <-r.Ready() + } + return p + } + if _, ok := seatDefinition(seatlist.NewPart("recovery", nil), meshbroker.Seat)(); ok { + t.Error("a definition is answered before the seat list's reader started") + } + if _, ok := seatDefinition(follow(true), meshbroker.Seat)(); ok { + t.Error("a definition is answered while the seat mesh-catalog has not answered the list") + } + read := follow(false) + if d, ok := seatDefinition(read, meshbroker.Seat)(); !ok || d.Seat != meshbroker.Seat { + t.Errorf("the read list's %s is answered as %q, %v", meshbroker.Seat, d.Seat, ok) + } + if _, ok := seatDefinition(read, "no-such-seat")(); ok { + t.Error("a seat the list does not hold is answered a definition") + } + }) t.Run("bus names of a kept data (A-121)", func(t *testing.T) { m, err := catalogue.Decode([]byte(`{"module":"app","version":"1","state":[{"name":"old"}],"logs":[{"name":"journal"}], "streams":[{"name":"APP_EVENTS"}],"objects":[{"name":"files"}], diff --git a/modules/mesh-recovery/cmd/mesh-recovery/whole_integration_test.go b/modules/mesh-recovery/cmd/mesh-recovery/whole_integration_test.go index 1ca0f260b..99b05d39c 100644 --- a/modules/mesh-recovery/cmd/mesh-recovery/whole_integration_test.go +++ b/modules/mesh-recovery/cmd/mesh-recovery/whole_integration_test.go @@ -32,6 +32,8 @@ import ( "git.novox.be/novox/mesh-sdk/go/buscalls" "git.novox.be/novox/mesh-sdk/go/bussnap" "git.novox.be/novox/mesh-sdk/go/process" + "git.novox.be/novox/mesh-sdk/go/seatdef" + "git.novox.be/novox/mesh-sdk/go/seatlist" "git.novox.be/novox/mesh-sdk/go/testbus" "mesh-recovery/internal/recovery" @@ -143,6 +145,30 @@ func TestRecoveryWhole(t *testing.T) { answer(m, map[string]any{"entry": map[string]any{"module": "app", "fingerprint": "src2:app1", "manifest": map[string]any{"module": "app", "version": "1", "state": []any{map[string]any{"name": "notes"}}}}}) }) + // The seat list (A-186): the seat mesh-catalog's seat-definitions names the module's seat, whose definition its verbs' + // callers are checked against: each verb of the contract, its class's default callers. + recoveryDef := seatdef.Definition{Format: seatdef.Format, Seat: seat.Seat, Scope: seatdef.ScopeMesh, Grain: seatdef.SeatList, + Decision: "novox/hq ADR 0672", Domain: "Core"} + for _, v := range seat.Verbs() { + recoveryDef.Verbs = append(recoveryDef.Verbs, seatdef.Verb{Name: v, Class: seatdef.ClassAct}) + } + rawDef, _ := json.Marshal(recoveryDef) + listed := make(chan struct{}) + var listedOnce sync.Once + sub("mesh.seat.mesh-catalog.call."+seatlist.Verb+".*", func(m *nats.Msg) { + var in struct { + Seat string `json:"seat"` + } + _ = json.Unmarshal(m.Data, &in) + entry := seatlist.Entry{Seat: seat.Seat, Source: seatlist.SourceTrunk, Digest: "sha256:recovery"} + if in.Seat == "" { + answer(m, seatlist.Listing{Revision: 1, Seats: []seatlist.Entry{entry}}) + return + } + entry.Definition = rawDef + answer(m, seatlist.Answer{Seat: in.Seat, Found: in.Seat == seat.Seat, Entry: &entry, Revision: 1}) + listedOnce.Do(func() { close(listed) }) + }) keeper := &standKeeper{} restored := func(request, act, outcome string) { b, _ := json.Marshal(meshbroker.SnapshotRestoredEvent{Request: request, Act: act, Outcome: outcome, Error: "cancelled"}) @@ -256,15 +282,20 @@ func TestRecoveryWhole(t *testing.T) { ask := func(t *testing.T, verb, caller string, body any) seat.SnapshotKeptAnswer { t.Helper() data, _ := json.Marshal(body) - m, err := nc.Request("mesh.seat.mesh-recovery.call."+verb+"."+strings.ReplaceAll(caller, ".", "~"), data, 10*time.Second) - if err != nil { - t.Fatalf("%s: %v", verb, err) - } var r struct { Result seat.SnapshotKeptAnswer `json:"result"` Error string `json:"error"` } - _ = json.Unmarshal(m.Data, &r) + // The definition was answered before the first ask (listed); the reader holds it as its read ends, so a refusal for + // an unread list is asked again, a bounded number of times. + for try := 0; try == 0 || strings.Contains(r.Error, "not yet read") && try < 100; try++ { + m, err := nc.Request("mesh.seat.mesh-recovery.call."+verb+"."+strings.ReplaceAll(caller, ".", "~"), data, 10*time.Second) + if err != nil { + t.Fatalf("%s: %v", verb, err) + } + r.Error = "" + _ = json.Unmarshal(m.Data, &r) + } if r.Error != "" { r.Result.Error = r.Error } @@ -313,6 +344,11 @@ func TestRecoveryWhole(t *testing.T) { waitEvent(t, "the start's removal of points past their keep days", func(h heard) bool { return h.subject == "mesh.seat.mesh-recovery.event."+seat.EventExpiryEnded }) + select { + case <-listed: + case <-time.After(30 * time.Second): + t.Fatal("the module did not read its seat's definition from the seat list within 30s") + } var first string t.Run("a hand take's log names the account the bus proved and the asker (§554)", func(t *testing.T) { diff --git a/modules/mesh-recovery/internal/recovery/manifest_test.go b/modules/mesh-recovery/internal/recovery/manifest_test.go index 3b8a7dd54..099ab70ed 100644 --- a/modules/mesh-recovery/internal/recovery/manifest_test.go +++ b/modules/mesh-recovery/internal/recovery/manifest_test.go @@ -130,7 +130,17 @@ func TestTheManifestSaysWhatTheModuleIs(t *testing.T) { if !bounds { t.Error("no state bounds with key-ttl for the expiry's bound key") } - for _, c := range []string{"node-backup.item-started", "node-backup.item-ended", "mesh-broker.snapshot-restored"} { + // The seat list its verbs' callers are checked against (A-186): read by its verb, by the process too, heard as it changes. + if !slices.Contains(m.Invokes, "seat:mesh-catalog.seat-definitions") { + t.Error("does not invoke seat:mesh-catalog.seat-definitions") + } + for _, r := range m.Resources { + if r.Type == "process" && !slices.Contains(r.Bus.Invokes, "seat:mesh-catalog.seat-definitions") { + t.Error("the process does not invoke seat:mesh-catalog.seat-definitions") + } + } + for _, c := range []string{"node-backup.item-started", "node-backup.item-ended", "mesh-broker.snapshot-restored", + "mesh-catalog.seat-definition-changed", "mesh-catalog.seat-definition-removed"} { if !slices.Contains(m.Consumes, c) { t.Errorf("does not consume %s", c) } diff --git a/modules/mesh-recovery/module.json b/modules/mesh-recovery/module.json index 179c4ed97..28978de32 100644 --- a/modules/mesh-recovery/module.json +++ b/modules/mesh-recovery/module.json @@ -37,9 +37,12 @@ "node-backup.item-ended", "node-engine.running", "mesh-broker.snapshot-restored", - "mesh-conditions.findings-asked" + "mesh-conditions.findings-asked", + "mesh-catalog.seat-definition-changed", + "mesh-catalog.seat-definition-removed" ], "invokes": [ + "seat:mesh-catalog.seat-definitions", "seat:mesh-catalog.entry", "seat:mesh-module-assigner.assignments", "seat:mesh-hand-acts.hand-act", @@ -260,6 +263,7 @@ ], "bus": { "invokes": [ + "seat:mesh-catalog.seat-definitions", "seat:mesh-catalog.entry", "seat:mesh-module-assigner.assignments", "seat:mesh-hand-acts.hand-act", diff --git a/modules/nats/cmd/nats-tools/keep.go b/modules/nats/cmd/nats-tools/keep.go index c5e1f0ba2..93afd23b7 100644 --- a/modules/nats/cmd/nats-tools/keep.go +++ b/modules/nats/cmd/nats-tools/keep.go @@ -242,7 +242,7 @@ func runKeeper(ctx context.Context, getenv func(string) string, logOut io.Writer } k.Catalog, k.Seats, k.Again = follow.view, follow.seats, follow.again // The process answers the seat mesh-broker's snapshot verbs itself (design 47 G §681, A-107), with the caller the bus proved. - stopAnswers, err := conn.AnswerSeat(meshbroker.Seat, snapshotAnswers(k, heldRequest(k.Store), keeperSnapshots(nc, logger.Printf))) + stopAnswers, err := conn.AnswerSeat(meshbroker.Seat, follow.definition(meshbroker.Seat), snapshotAnswers(k, heldRequest(k.Store), keeperSnapshots(nc, logger.Printf))) if err != nil { return fmt.Errorf("the seat %s's snapshot verbs cannot be answered: %w", meshbroker.Seat, err) } @@ -318,6 +318,15 @@ func (f *catalogFollow) seats() map[string]seatdef.Definition { return f.seatList.Definitions() } +// definition is the definition of seat the seat list holds, which the seat's verbs answered by the process check their callers +// against (process.Conn.AnswerSeat; design 47 G lead answers A-179, A-186): none until the seat has answered, as seats says. +func (f *catalogFollow) definition(seat string) func() (seatdef.Definition, bool) { + return func() (seatdef.Definition, bool) { + d, ok := f.seats()[seat] + return d, ok + } +} + // again asks every first view of the catalog half and the seat list again, by hand. func (f *catalogFollow) again() error { var errs []error diff --git a/modules/nats/cmd/nats-tools/keep_test.go b/modules/nats/cmd/nats-tools/keep_test.go index a8104d208..d42ca1858 100644 --- a/modules/nats/cmd/nats-tools/keep_test.go +++ b/modules/nats/cmd/nats-tools/keep_test.go @@ -1,11 +1,17 @@ package main import ( + "encoding/json" + "errors" "fmt" "strings" "testing" "time" + "git.novox.be/novox/mesh-contracts/go/seats/meshbroker" + "git.novox.be/novox/mesh-sdk/go/busretry" + "git.novox.be/novox/mesh-sdk/go/seatlist" + catalog "git.novox.be/novox/mesh-delivery/modules/mesh-delivery-contracts/go/seats/meshcatalog" "nats-tools/internal/keeper" @@ -99,3 +105,43 @@ func TestTheKeepersHealthToolReadsItsStatusAndItsConsumer(t *testing.T) { }) } } + +// listingCatalog is the seat mesh-catalog as the reader asks it: an empty listing, so the bootstrapped definitions stand, or, +// silent, no answer at all. +type listingCatalog struct{ silent bool } + +func (c listingCatalog) Ask(string, any) (json.RawMessage, error) { + if c.silent { + return nil, errors.New("no responders") + } + return json.Marshal(seatlist.Listing{Revision: 1}) +} + +func (listingCatalog) Subscribe(string, func(string, json.RawMessage) error) error { return nil } + +// A-179, A-186: the definition the seat mesh-broker's snapshot verbs check their callers against is the seat list's, none +// before the reader started or while the seat mesh-catalog has not answered, and none for a seat the list does not hold. +func TestTheAnsweredSeatsDefinitionIsTheSeatListsOnceRead(t *testing.T) { + t.Cleanup(busretry.Waits(20*time.Millisecond, time.Millisecond, 5*time.Millisecond)) + follow := func(silent bool) *catalogFollow { + r, err := seatlist.Start(seatlist.Config{Bus: listingCatalog{silent: silent}, Bootstrap: seatlist.Bootstrapped()}) + if err != nil { + t.Fatal(err) + } + <-r.Ready() + return &catalogFollow{seatList: r} + } + if _, ok := (&catalogFollow{}).definition(meshbroker.Seat)(); ok { + t.Error("a definition is answered with no seat list") + } + if _, ok := follow(true).definition(meshbroker.Seat)(); ok { + t.Error("a definition is answered while the seat mesh-catalog has not answered the list") + } + read := follow(false) + if d, ok := read.definition(meshbroker.Seat)(); !ok || d.Seat != meshbroker.Seat { + t.Errorf("the read list's %s is answered as %q, %v", meshbroker.Seat, d.Seat, ok) + } + if _, ok := read.definition("no-such-seat")(); ok { + t.Error("a seat the list does not hold is answered a definition") + } +} diff --git a/modules/nats/cmd/nats-tools/keeper_integration_test.go b/modules/nats/cmd/nats-tools/keeper_integration_test.go index 2744d2f45..2f7e0f14b 100644 --- a/modules/nats/cmd/nats-tools/keeper_integration_test.go +++ b/modules/nats/cmd/nats-tools/keeper_integration_test.go @@ -29,6 +29,7 @@ import ( "git.novox.be/novox/mesh-sdk/go/catalogue" "git.novox.be/novox/mesh-sdk/go/process" "git.novox.be/novox/mesh-sdk/go/seatdef" + "git.novox.be/novox/mesh-sdk/go/seatlist" "git.novox.be/novox/mesh-sdk/go/views" "git.novox.be/novox/mesh-sdk/go/bussnap" @@ -467,7 +468,16 @@ func TestTheBusModuleRunsWhole(t *testing.T) { t.Fatalf("the drop's record holds no snapshot: %+v", rec) } world := keeperSnapshots(nc, t.Logf) - stopAnswering, err := pconn.AnswerSeat(meshbroker.Seat, snapshotAnswers(k, heldRequest(store), world)) + // The seat's definition as the seat list holds it, the bootstrapped one (A-186). + broker := func() (seatdef.Definition, bool) { + for _, d := range seatlist.Bootstrapped() { + if d.Seat == meshbroker.Seat { + return d, true + } + } + return seatdef.Definition{}, false + } + stopAnswering, err := pconn.AnswerSeat(meshbroker.Seat, broker, snapshotAnswers(k, heldRequest(store), world)) if err != nil { t.Fatal(err) }