diff --git a/internal/agent/kept_test.go b/internal/agent/kept_test.go index 144c5f86..9578891b 100644 --- a/internal/agent/kept_test.go +++ b/internal/agent/kept_test.go @@ -53,7 +53,9 @@ func (f *recoveryFake) next(verb string, in any) (string, string, string) { func (f *recoveryFake) env() *Recovery { return &Recovery{ - FormChanges: func(context.Context, string, string, string) ([]catalogue.FormChange, error) { return f.changes, f.unknown }, + FormChanges: func(context.Context, string, string, string) ([]catalogue.FormChange, error) { + return f.changes, f.unknown + }, SnapshotKept: func(_ context.Context, in meshrecovery.SnapshotKeptInput) (meshrecovery.SnapshotKeptAnswer, error) { a, b, e := f.next("snapshot-kept", in) return meshrecovery.SnapshotKeptAnswer{Accepted: a, Busy: b, Error: e}, nil diff --git a/internal/link/engineseat.go b/internal/link/engineseat.go index 88e01c1f..f63404b1 100644 --- a/internal/link/engineseat.go +++ b/internal/link/engineseat.go @@ -236,14 +236,10 @@ func sayOnce(publish func(id string) error, subject string, raw []byte) error { // 0563 §3 changes it): its ask and cancel on the seat's work stream, its warrants heard on its own decided subject, and // its asks' state read from the router's records under its own name. -// RouterRecords is the router's bucket of asks, which each asker reads under its own name: the state `asks` of the -// operator-channel's holder, the messenger (mesh-catalog modules/messenger). -const RouterRecords = "messenger_asks" - // OperatorAsks is a link that asks the operator as this node's engine. Optional. type OperatorAsks interface { PublishAsk(subject string, body any, id string) error - RouterRecord(key string) (*stdio.StateEntry, error) + RouterRecord(bucket, key string) (*stdio.StateEntry, error) HearWarrants(asker string, heard func(oc.Warrant)) error } @@ -259,11 +255,12 @@ func (l *natsLink) PublishAsk(subject string, body any, id string) error { return err } -// RouterRecord reads one key of the router's records, `.`; nil when it holds none. -func (l *natsLink) RouterRecord(key string) (*stdio.StateEntry, error) { +// RouterRecord reads one key of the router's records, `.`, in the bucket the seat list names (mesh-sdk's operatorask); nil +// when it holds none. +func (l *natsLink) RouterRecord(bucket, key string) (*stdio.StateEntry, error) { ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second) defer cancel() - msg, err := l.conn.RequestWithContext(ctx, "$JS.API.DIRECT.GET.KV_"+RouterRecords+".$KV."+RouterRecords+"."+key, nil) + msg, err := l.conn.RequestWithContext(ctx, "$JS.API.DIRECT.GET.KV_"+bucket+".$KV."+bucket+"."+key, nil) if err != nil { return nil, err } diff --git a/internal/link/have_test.go b/internal/link/have_test.go index 4c2ceb52..084557bf 100644 --- a/internal/link/have_test.go +++ b/internal/link/have_test.go @@ -7,6 +7,8 @@ import ( "fmt" "git.novox.be/novox/mesh-contracts/go/clients/nodeengineclient" nodeengine "git.novox.be/novox/mesh-contracts/go/seats/nodeengine" + oc "git.novox.be/novox/mesh-contracts/operator-and-conversation/go/seats/operatorchannel" + stdio "git.novox.be/novox/mesh-sdk/go" sdkkeys "git.novox.be/novox/mesh-sdk/go/keys" "github.com/novox/mesh-host/internal/compositions" "slices" @@ -46,6 +48,8 @@ type haveLink struct { said []string // onSaid are the events said on a subject of their own (running), by subject, in order. onSaid []string + // asked are the subjects asked of the bus, records the router's records read as bucket and key. + asked, records []string } func (l *haveLink) SayEngineOn(subject string, body any) error { @@ -97,6 +101,23 @@ type haveEntry struct { rev uint64 } +// The link as the operator's asker and a tool bus: the seat list answers the holder of the seat operator-channel, and the +// router's records are read from the bucket named for it (design 47 G A-101). +func (l *haveLink) Ask(_ context.Context, subject string, _ []byte) ([]byte, error) { + l.hmu.Lock() + l.asked = append(l.asked, subject) + l.hmu.Unlock() + return []byte(`{"result":{"seats":[{"seat":"operator-channel","holders":[{"module":"the-router"}]}]}}`), nil +} +func (l *haveLink) PublishAsk(string, any, string) error { return nil } +func (l *haveLink) HearWarrants(string, func(oc.Warrant)) error { return nil } +func (l *haveLink) RouterRecord(bucket, key string) (*stdio.StateEntry, error) { + l.hmu.Lock() + l.records = append(l.records, bucket+" "+key) + l.hmu.Unlock() + return nil, nil +} + func newHaveLink() *haveLink { return &haveLink{quietLink: newQuietLink(), kv: map[string]haveEntry{}} } func (l *haveLink) Health(context.Context, string, []byte) error { return nil } @@ -715,6 +736,11 @@ func TestAnAppliedModulesHaveNamesTheFingerprintAndTheDigestAndFollowsItsHealth( if got := l.saidOn("mesh.seat.node-engine.event.wait-ended.n1.web "); len(got) != 1 { t.Fatalf("wait-ended said %v, want one", got) } + // The router's record of an ask is read from the bucket of the seat's holder on the seat list (design 47 G A-101). + if _, err := q.RouterRecord("engine.a1"); err != nil || !slices.Equal(l.records, []string{"the-router_asks engine.a1"}) || + !strings.Contains(strings.Join(l.asked, " "), "assignments") { + t.Fatalf("the router's record read as %v after asking %v: %v", l.records, l.asked, err) + } // Starting: it waits. Past its start window the check fails: failed. q.SayHealth(context.Background(), *healthOf(map[string]string{"web": StateStarting})) if got := l.have(t, "n1.web"); got.Health != nodeengineclient.HealthWaits || got.Fp != fp || got.Composed != dig { diff --git a/internal/link/session.go b/internal/link/session.go index 59bfb062..8bb427f5 100644 --- a/internal/link/session.go +++ b/internal/link/session.go @@ -12,9 +12,12 @@ import ( "sync" "time" + "git.novox.be/novox/mesh-contracts/go/seats/meshmoduleassigner" "git.novox.be/novox/mesh-contracts/go/seats/nodeengine" + occ "git.novox.be/novox/mesh-contracts/operator-and-conversation/go/seats/operatorchannel" stdio "git.novox.be/novox/mesh-sdk/go" "git.novox.be/novox/mesh-sdk/go/health" + "git.novox.be/novox/mesh-sdk/go/operatorask" ) // Session is the node-engine's hold on the bus (novox/hq ADR 0539 §9): the link open now, what is asked of this @@ -403,7 +406,18 @@ func (q *Session) RouterRecord(key string) (*stdio.StateEntry, error) { if err != nil { return nil, err } - return asking.RouterRecord(key) + // The bucket is the one of the seat's holder on the seat list, found by mesh-sdk's operatorask (design 47 G A-101). + bucket, err := operatorask.Bucket(func(call string, body any) (json.RawMessage, error) { + ctx, cancel := context.WithTimeout(context.Background(), VaultWithin) + defer cancel() + return q.AskMeshSeat(ctx, meshmoduleassigner.Seat, meshmoduleassigner.VerbAssignments, body) + }, operatorask.Contract{Lists: "seat:" + meshmoduleassigner.Seat + "." + meshmoduleassigner.VerbAssignments, + ListBody: meshmoduleassigner.AssignmentsInput{Seats: meshmoduleassigner.AssignmentsInputSeatsTrue}, + Seat: occ.Seat, StateAsks: occ.StateAsks}) + if err != nil { + return nil, err + } + return asking.RouterRecord(bucket, key) } // StatedReads is what this engine states it reads of a manifest and when it made that statement, at its start or at the diff --git a/module.json b/module.json index 29d7ff97..c66f253f 100644 --- a/module.json +++ b/module.json @@ -42,6 +42,7 @@ "invokes": [ "seat:mesh-catalog.entry", "seat:mesh-catalog.seat-definitions", + "seat:mesh-module-assigner.assignments", "seat:mesh-recovery.kept-points", "seat:mesh-recovery.restore-kept", "seat:mesh-recovery.snapshot-kept"