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
This commit is contained in:
@@ -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
|
||||
}
|
||||
}
|
||||
|
||||
@@ -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"}],
|
||||
|
||||
@@ -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) {
|
||||
|
||||
@@ -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)
|
||||
}
|
||||
|
||||
@@ -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",
|
||||
|
||||
@@ -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
|
||||
|
||||
@@ -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")
|
||||
}
|
||||
}
|
||||
|
||||
@@ -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)
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user