indexer: the router's record of an ask is read through the seat operator-channel's state asks, under the indexer's own name
routerRecordUnread is replaced by routerRecordOf: operatorchannelclient's Record is a get of the key (<module>.<id>) on the process's own bus connection (process.Conn.Ask), in the bucket formed from the seat's holder on the seat list (the seat mesh-module-assigner's verb assignments with seats) and the contract's StateAsks, no module's name in the program. The manifest uses operator-channel and invokes that verb. Builds: design 47 G §700, A-74 Claude-Session: https://claude.ai/code/session_01NcbgfGKaxw984XyAPfLm4h
This commit is contained in:
@@ -452,7 +452,7 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro
|
||||
registry := newRegistry(js, conn, cred, calls.Call,
|
||||
awaitedStore{b: process.AwaitBucket(ctx, js, conn, index.RegistrationsBucket, "no registration is kept or written", logger.Printf)},
|
||||
awaitedStore{b: process.AwaitBucket(ctx, js, conn, index.ProposalsBucket, "no change of a registration waits for its warrant", logger.Printf)},
|
||||
routerRecordUnread, ix, logger.Printf)
|
||||
routerRecordOf(cred.Module, calls.Call, calls.Ask), ix, logger.Printf)
|
||||
ix.Registry = registry
|
||||
named.Changed = ix.RemoveUnlisted
|
||||
ix.SeedModules = func(context.Context) ([]index.SeedModule, error) {
|
||||
|
||||
@@ -3,14 +3,16 @@ package main
|
||||
import (
|
||||
"context"
|
||||
"encoding/json"
|
||||
"errors"
|
||||
"fmt"
|
||||
"strings"
|
||||
"time"
|
||||
|
||||
mma "git.novox.be/novox/mesh-contracts/go/seats/meshmoduleassigner"
|
||||
"git.novox.be/novox/mesh-contracts/operator-and-conversation/go/clients/operatorchannelclient"
|
||||
oc "git.novox.be/novox/mesh-contracts/operator-and-conversation/go/seats/operatorchannel"
|
||||
gitseat "git.novox.be/novox/mesh-delivery/contracts/go/seats/git"
|
||||
"git.novox.be/novox/mesh-delivery/contracts/go/seats/meshcatalog"
|
||||
stdio "git.novox.be/novox/mesh-sdk/go"
|
||||
"git.novox.be/novox/mesh-sdk/go/catalogue"
|
||||
"git.novox.be/novox/mesh-sdk/go/process"
|
||||
"github.com/nats-io/nats.go"
|
||||
@@ -47,7 +49,7 @@ func historySubject(module string) string {
|
||||
}
|
||||
|
||||
// newRegistry is the Registry over the indexer's own states, the stream registration-history, the seat git, and the seat
|
||||
// operator-channel's accepts. where is the read of the router's own record of an ask (design question D-37, open).
|
||||
// operator-channel's accepts. where is the read of the router's own record of an ask (routerRecordOf, §700).
|
||||
func newRegistry(js jetstream.JetStream, conn *nats.Conn, cred process.Account, call toolCaller, registrations, proposals index.Store, where func(id string) (string, *oc.Warrant, error), ix *index.Indexer, logf func(string, ...any)) *index.Registry {
|
||||
return &index.Registry{Registrations: registrations, Proposals: proposals,
|
||||
History: func(ctx context.Context, r meshcatalog.Registration) error {
|
||||
@@ -122,11 +124,42 @@ func newRegistry(js jetstream.JetStream, conn *nats.Conn, cred process.Account,
|
||||
Logf: logf}
|
||||
}
|
||||
|
||||
// errRouterRecord is what the read of the router's record answers until the design says how this program reads it (D-37): a
|
||||
// warrant is then left unapplied and said, never applied on the decided event's body alone (design 47 G §667, Heard).
|
||||
var errRouterRecord = errors.New("the router's own record of an ask is not readable from the indexer's program yet (design question D-37)")
|
||||
// routerBucket is the bucket of the router's asks: the state `asks` of the seat operator-channel (the contract's StateAsks) in
|
||||
// the bucket of the seat's holder on the seat list, which the seat mesh-module-assigner's verb assignments answers with seats
|
||||
// (design 47 G §700, A-74); no module's name is in the program.
|
||||
func routerBucket(call toolCaller) (string, error) {
|
||||
raw, err := call("seat:"+mma.Seat+"."+mma.VerbAssignments, mma.AssignmentsInput{Seats: mma.AssignmentsInputSeatsTrue})
|
||||
if err != nil {
|
||||
return "", fmt.Errorf("who holds the seat %s is not asked of the seat %s: %w", oc.Seat, mma.Seat, err)
|
||||
}
|
||||
var held struct {
|
||||
Seats []mma.SeatHolders `json:"seats"`
|
||||
}
|
||||
if err := json.Unmarshal(raw, &held); err != nil {
|
||||
return "", fmt.Errorf("the answer of %s.%s is not its seat's: %w", mma.Seat, mma.VerbAssignments, err)
|
||||
}
|
||||
for _, s := range held.Seats {
|
||||
if s.Seat == oc.Seat && len(s.Holders) > 0 {
|
||||
return s.Holders[0].Module + "_" + oc.StateAsks, nil
|
||||
}
|
||||
}
|
||||
return "", fmt.Errorf("no module holds the seat %s on the seat list, so the router's record of an ask is not read", oc.Seat)
|
||||
}
|
||||
|
||||
func routerRecordUnread(string) (string, *oc.Warrant, error) { return "", nil, errRouterRecord }
|
||||
// routerRecordOf is Registry.Where: the router's own record of an ask, read under the indexer's module's name through
|
||||
// operatorchannelclient, whose Record is a get of that key (`<module>.<id>` of the seat's state asks) on the process's own bus
|
||||
// connection (record is process.Conn.Ask), as the tool runner gives it the runtime's mesh/seat.record (design 47 G §700): never
|
||||
// the decided event's body. A warrant is then left unapplied and said while the record cannot be read.
|
||||
func routerRecordOf(module string, call toolCaller, record func(bucket, id string) (*stdio.StateEntry, error)) func(id string) (string, *oc.Warrant, error) {
|
||||
client := &operatorchannelclient.Client{Module: module, Record: func(id string) (*stdio.StateEntry, error) {
|
||||
bucket, err := routerBucket(call)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
return record(bucket, id)
|
||||
}}
|
||||
return client.Where
|
||||
}
|
||||
|
||||
// heardDecided is the handler of the decided events for this module: a warrant is applied by the registry; one it could not
|
||||
// apply is left unacknowledged, so the bus gives it again, and said.
|
||||
|
||||
@@ -0,0 +1,37 @@
|
||||
package main
|
||||
|
||||
import (
|
||||
"encoding/json"
|
||||
"strings"
|
||||
"testing"
|
||||
|
||||
stdio "git.novox.be/novox/mesh-sdk/go"
|
||||
)
|
||||
|
||||
// Design 47 G §700, A-74: the router's record of an ask is read under the indexer's module's name from the state asks of the
|
||||
// seat operator-channel, in the bucket of the seat's holder on the seat list, and carries the ask's warrant.
|
||||
func TestTheRoutersRecordIsReadFromTheBucketOfTheSeatsHolder(t *testing.T) {
|
||||
var asked []string
|
||||
call := func(key string, body any) (json.RawMessage, error) {
|
||||
asked = append(asked, key)
|
||||
return json.RawMessage(`{"seats":[{"seat":"node-packet-filter","scope":"node","node":"n1","holders":[{"module":"pf","node":"n1"}]},` +
|
||||
`{"seat":"operator-channel","scope":"mesh","holders":[{"module":"ops-router","node":"n2"}]}],"holds":2}`), nil
|
||||
}
|
||||
var bucket, id string
|
||||
record := func(b, i string) (*stdio.StateEntry, error) {
|
||||
bucket, id = b, i
|
||||
return &stdio.StateEntry{Key: "mesh-delivery-indexer." + i, Value: json.RawMessage(`{"state":"chosen","warrant":{"ask":"a1","asker":"mesh-delivery-indexer","choice":"approve"}}`)}, nil
|
||||
}
|
||||
state, w, err := routerRecordOf("mesh-delivery-indexer", call, record)("a1")
|
||||
if err != nil || state != "chosen" || w == nil || w.Ask != "a1" {
|
||||
t.Fatalf("Where = %q, %+v, %v; want the state chosen and the warrant of a1", state, w, err)
|
||||
}
|
||||
if bucket != "ops-router_asks" || id != "a1" || len(asked) != 1 || asked[0] != "seat:mesh-module-assigner.assignments" {
|
||||
t.Errorf("read %q key %q after asking %v; want the holder's bucket ops-router_asks, the id, one question of the seat mesh-module-assigner", bucket, id, asked)
|
||||
}
|
||||
// No holder on the seat list: nothing is read, and the warrant stays unapplied with the reason said.
|
||||
none := func(string, any) (json.RawMessage, error) { return json.RawMessage(`{"seats":[],"holds":0}`), nil }
|
||||
if _, _, err := routerRecordOf("mesh-delivery-indexer", none, record)("a1"); err == nil || !strings.Contains(err.Error(), "no module holds the seat operator-channel") {
|
||||
t.Errorf("with no holder Where answered %v; want the reason", err)
|
||||
}
|
||||
}
|
||||
@@ -75,6 +75,7 @@ type manifest struct {
|
||||
State []state `json:"state"`
|
||||
Reads []string `json:"reads"`
|
||||
Invokes []string `json:"invokes"`
|
||||
Uses []string `json:"uses"`
|
||||
OwnSecrets map[string]string `json:"own-secrets"`
|
||||
SecretsOwner string `json:"secrets-owner"`
|
||||
Resources []map[string]any `json:"resources"`
|
||||
@@ -119,6 +120,11 @@ func TestManifestDeclarations(t *testing.T) { // 1238/AC1, 1444 (reads the build
|
||||
if own := slices.DeleteFunc(slices.Clone(m.Emits), func(e string) bool { return strings.HasPrefix(e, views.EventPrefix) }); !reflect.DeepEqual(own, []string{takein.TakeInPendingEvent, takein.TakeInTakenEvent}) {
|
||||
t.Errorf("emits = %v, want the two take-in events (the verdict events are the seat mesh-catalog's)", m.Emits)
|
||||
}
|
||||
// Design 47 G §700, A-74: the router's record of an ask is read through the seat operator-channel, which the indexer uses,
|
||||
// and its holder is asked of the seat mesh-module-assigner.
|
||||
if !slices.Equal(m.Uses, []string{"operator-channel"}) || !slices.Contains(m.Invokes, "seat:mesh-module-assigner.assignments") {
|
||||
t.Errorf("uses = %v, invokes = %v, want the seat operator-channel used and seat:mesh-module-assigner.assignments invoked", m.Uses, m.Invokes)
|
||||
}
|
||||
// 1126 note 76: heard keeps the last 64 FORGE messages the indexer handled, for indexer_heard.
|
||||
if want := []state{{Name: "candidates", History: 10}, {Name: "heard", History: 64}, {Name: "manifests", History: 1}, {Name: "predefined", History: 5}, {Name: "config-candidates", History: 5}, {Name: "taken", History: 1}, {Name: "takeins", History: 1}, {Name: "verdicts", History: 5}, {Name: "oldsections", History: 1}, {Name: "try-recovery", History: 1}, {Name: "own-build", History: 1}, {Name: index.PackagesState, History: index.PackagesHistory},
|
||||
{Name: views.MarksState, History: 1}, {Name: index.FindingsState, History: 5}, {Name: views.StatusState, History: 1},
|
||||
@@ -172,7 +178,7 @@ func TestManifestDeclarations(t *testing.T) { // 1238/AC1, 1444 (reads the build
|
||||
// Three invokes: the file of a registered module's manifest and lock files (registered.go), the forge's trees (the named paths' objects, since the indexer is the one reader of a manifest's paths:
|
||||
// ADR 0499 clauses 2 and 3, ADR 0495 clause 1; and the module directories of the trunk, issue 1900) and whether a
|
||||
// commit is in another (issue 1900).
|
||||
if want := []string{"mesh-delivery-assigner.assignments", "seat:git.trees", "seat:git.file", "seat:git.contains", "seat:git.open-pulls", "seat:git.branch", "mesh-delivery-checker.checker_tries", "mesh-delivery-forge-bridge.bridge_merged_message", "seat:node-engine.composition", "seat:mesh-delivery.source", "seat:mesh-hand-acts.hand-act", "seat:mesh-watchdogs.presence"}; !reflect.DeepEqual(m.Invokes, want) {
|
||||
if want := []string{"mesh-delivery-assigner.assignments", "seat:git.trees", "seat:git.file", "seat:git.contains", "seat:git.open-pulls", "seat:git.branch", "mesh-delivery-checker.checker_tries", "mesh-delivery-forge-bridge.bridge_merged_message", "seat:node-engine.composition", "seat:mesh-delivery.source", "seat:mesh-module-assigner.assignments", "seat:mesh-hand-acts.hand-act", "seat:mesh-watchdogs.presence"}; !reflect.DeepEqual(m.Invokes, want) {
|
||||
t.Errorf("invokes = %v, want %v", m.Invokes, want)
|
||||
}
|
||||
// 1945: indexer_take_in, served by its tools bundle through the takeins state; the process serves none. The same bundle
|
||||
@@ -312,8 +318,8 @@ func TestManifestDeclarations(t *testing.T) { // 1238/AC1, 1444 (reads the build
|
||||
if !slices.Contains(m.Emits, takein.TakeInPendingEvent) || !slices.Contains(m.Emits, takein.TakeInTakenEvent) {
|
||||
t.Errorf("emits = %v: want both take-in events emitted", m.Emits)
|
||||
}
|
||||
// No bus, no other seat (ADR 0437 §4).
|
||||
for _, k := range []string{"logs", "provides", "requires", "uses", "defines-seats", "seat", "seats", "bus"} {
|
||||
// No bus, no other seat (ADR 0437 §4); it uses the seat operator-channel alone (design 47 G §700, A-74, above).
|
||||
for _, k := range []string{"logs", "provides", "requires", "defines-seats", "seat", "seats", "bus"} {
|
||||
if _, ok := top[k]; ok {
|
||||
t.Errorf("the manifest declares %q: the indexer has none", k)
|
||||
}
|
||||
|
||||
@@ -152,9 +152,13 @@
|
||||
"mesh-delivery-forge-bridge.bridge_merged_message",
|
||||
"seat:node-engine.composition",
|
||||
"seat:mesh-delivery.source",
|
||||
"seat:mesh-module-assigner.assignments",
|
||||
"seat:mesh-hand-acts.hand-act",
|
||||
"seat:mesh-watchdogs.presence"
|
||||
],
|
||||
"uses": [
|
||||
"operator-channel"
|
||||
],
|
||||
"own-secrets": {
|
||||
"broker": "${dir:state}/broker"
|
||||
},
|
||||
|
||||
Reference in New Issue
Block a user