A-101: the node-engine reads the router's record through mesh-sdk's operatorask, the bucket from the seat list
Builds: design 47 G A-101, §700, A-74 Claude-Session: https://claude.ai/code/session_01NcbgfGKaxw984XyAPfLm4h
This commit is contained in:
@@ -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
|
||||
|
||||
@@ -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, `<asker>.<id>`; nil when it holds none.
|
||||
func (l *natsLink) RouterRecord(key string) (*stdio.StateEntry, error) {
|
||||
// RouterRecord reads one key of the router's records, `<asker>.<id>`, 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
|
||||
}
|
||||
|
||||
@@ -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 {
|
||||
|
||||
@@ -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
|
||||
|
||||
@@ -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"
|
||||
|
||||
Reference in New Issue
Block a user