Merge branch 'feat/47-G-fix9-delivery' into feat/47-G-join9
The view's holderOf keeps the seatlist.HolderOf form of builder2 (A-126) over fix9-delivery's own assignments read. Claude-Session: https://claude.ai/code/session_01NcbgfGKaxw984XyAPfLm4h
This commit is contained in:
@@ -86,9 +86,11 @@ accounts {
|
||||
"$JS.ACK.EVENTS.n1_mesh-delivery-forge-bridge.>"
|
||||
"$JS.API.CONSUMER.INFO.EVENTS.n1_mesh-delivery-forge-bridge"
|
||||
"$JS.API.CONSUMER.MSG.NEXT.EVENTS.n1_mesh-delivery-forge-bridge"
|
||||
"$JS.API.STREAM.INFO.SEAT_mesh-delivery_owed"
|
||||
"$JS.ACK.SEAT_mesh-delivery_owed.n1_mesh-delivery-forge-bridge.>"
|
||||
"$JS.API.CONSUMER.INFO.SEAT_mesh-delivery_owed.n1_mesh-delivery-forge-bridge"
|
||||
"$JS.API.CONSUMER.MSG.NEXT.SEAT_mesh-delivery_owed.n1_mesh-delivery-forge-bridge"
|
||||
"$JS.API.STREAM.INFO.SEAT_mesh-delivery-checker_owed"
|
||||
"$JS.ACK.SEAT_mesh-delivery-checker_owed.n1_mesh-delivery-forge-bridge.>"
|
||||
"$JS.API.CONSUMER.INFO.SEAT_mesh-delivery-checker_owed.n1_mesh-delivery-forge-bridge"
|
||||
"$JS.API.CONSUMER.MSG.NEXT.SEAT_mesh-delivery-checker_owed.n1_mesh-delivery-forge-bridge"
|
||||
|
||||
@@ -8,12 +8,12 @@ import (
|
||||
"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/operatorask"
|
||||
"git.novox.be/novox/mesh-sdk/go/process"
|
||||
"github.com/nats-io/nats.go"
|
||||
"github.com/nats-io/nats.go/jetstream"
|
||||
@@ -135,41 +135,16 @@ func newRegistry(js jetstream.JetStream, conn *nats.Conn, cred process.Account,
|
||||
Logf: logf}
|
||||
}
|
||||
|
||||
// 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)
|
||||
}
|
||||
|
||||
// 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.
|
||||
// routerRecordOf is Registry.Where: the router's own record of an ask, read under the indexer's module's name by mesh-sdk's
|
||||
// operatorask (design 47 G A-101, §700, A-74): 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;
|
||||
// record is process.Conn.Ask, as the tool runner gives it the runtime's mesh/seat.record: 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
|
||||
asker := &operatorask.Asker[oc.Warrant]{Module: module, Call: operatorask.Caller(call), Record: record,
|
||||
Contract: operatorask.Contract{Lists: "seat:" + mma.Seat + "." + mma.VerbAssignments, ListBody: mma.AssignmentsInput{Seats: mma.AssignmentsInputSeatsTrue},
|
||||
Seat: oc.Seat, StateAsks: oc.StateAsks}}
|
||||
return asker.Where
|
||||
}
|
||||
|
||||
// heardDecided is the handler of the decided events for this module: a warrant is applied by the registry; one it could not
|
||||
|
||||
@@ -2457,11 +2457,11 @@ func caseAProcessBusNamesOnlyTheModulesOwn(t *testing.T) {
|
||||
}
|
||||
}
|
||||
|
||||
// 1179/AC4: a bus declaring serves, holds, claims or a seat is refused naming the key; a bus with invokes of
|
||||
// a tool or a seat verb is accepted.
|
||||
// 1179/AC4, changed by design 47 G A-107, A-109, A-110: a bus declaring holds, claims or a seat is refused naming the
|
||||
// key; a bus declaring serves (the seat verbs the process answers, each seat:<seat>.<verb>) is accepted and given whole to
|
||||
// the grant composition, as is a bus with invokes of a tool or a seat verb.
|
||||
func caseAProcessBusNeverServesOrHoldsASeat(t *testing.T) {
|
||||
for name, c := range map[string]struct{ bus, says string }{
|
||||
"serves": {`{"emits":["mail.send.sent"],"serves":["send"]}`, "serves"},
|
||||
"holds": {`{"emits":["mail.send.sent"],"holds":["mail-sender"]}`, "holds"},
|
||||
"claims": {`{"emits":["mail.send.sent"],"claims":["mail-sender"]}`, "claims"},
|
||||
"seat": {`{"emits":["mail.send.sent"],"seat":"mail-sender"}`, "seat"},
|
||||
@@ -2475,6 +2475,14 @@ func caseAProcessBusNeverServesOrHoldsASeat(t *testing.T) {
|
||||
if _, err := ParseManifest([]byte(mailSenderProcessManifest(t, accepted))); err != nil {
|
||||
t.Fatalf("a bus invoking a tool and a seat verb was refused: %v", err)
|
||||
}
|
||||
serving := `{"emits":["mail.send.sent"],"serves":["seat:mail-sender.send"]}`
|
||||
m, err := ParseManifest([]byte(mailSenderProcessManifest(t, serving)))
|
||||
if err != nil {
|
||||
t.Fatalf("a bus serving a seat verb was refused: %v", err)
|
||||
}
|
||||
if got := m.ProcessBuses()["send"].Serves; !strSliceEqual(got, []string{"seat:mail-sender.send"}) {
|
||||
t.Fatalf("the process's serves read as %v", got)
|
||||
}
|
||||
}
|
||||
|
||||
// 1179/AC1: a well-formed bus on a process whose account the module makes is accepted, and the catalogue
|
||||
|
||||
@@ -3,16 +3,15 @@ package main
|
||||
import (
|
||||
"encoding/json"
|
||||
"errors"
|
||||
"fmt"
|
||||
"time"
|
||||
|
||||
"github.com/nats-io/nats.go"
|
||||
"github.com/nats-io/nats.go/jetstream"
|
||||
|
||||
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"
|
||||
stdio "git.novox.be/novox/mesh-sdk/go"
|
||||
"git.novox.be/novox/mesh-sdk/go/operatorask"
|
||||
"git.novox.be/novox/mesh-sdk/go/process"
|
||||
|
||||
"mesh-delivery-pipeline/internal/pipeline"
|
||||
@@ -21,40 +20,15 @@ import (
|
||||
// toolCaller is a kept call of a seat's verb on the bus (process.Conn.Call).
|
||||
type toolCaller func(key string, body any) (json.RawMessage, error)
|
||||
|
||||
// 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. (The indexer's function, to move into mesh-sdk.)
|
||||
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)
|
||||
}
|
||||
|
||||
// routerRecordOf is Hearer.Where: the router's own record of an ask, read under the runner'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 (design 47 G §700, A-89): never the decided event's body.
|
||||
// routerRecordOf is Hearer.Where: the router's own record of an ask, read under the runner's module's name by mesh-sdk's
|
||||
// operatorask (design 47 G A-101, §700, A-89): the state asks of the seat operator-channel in the bucket of the seat's holder
|
||||
// on the seat list, which the seat mesh-module-assigner's verb assignments answers with seats; record is a get of the key
|
||||
// `<module>.<id>` on the process's own bus connection: never the decided event's body.
|
||||
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
|
||||
asker := &operatorask.Asker[oc.Warrant]{Module: module, Call: operatorask.Caller(call), Record: record,
|
||||
Contract: operatorask.Contract{Lists: "seat:" + mma.Seat + "." + mma.VerbAssignments, ListBody: mma.AssignmentsInput{Seats: mma.AssignmentsInputSeatsTrue},
|
||||
Seat: oc.Seat, StateAsks: oc.StateAsks}}
|
||||
return asker.Where
|
||||
}
|
||||
|
||||
// busAsker publishes the runner's approval ask on the seat operator-channel's accept subject under the runner's own name, its
|
||||
|
||||
@@ -4,15 +4,15 @@ import (
|
||||
"context"
|
||||
"encoding/json"
|
||||
"errors"
|
||||
mma "git.novox.be/novox/mesh-contracts/go/seats/meshmoduleassigner"
|
||||
occ "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"
|
||||
"strings"
|
||||
"time"
|
||||
|
||||
"github.com/nats-io/nats.go"
|
||||
"github.com/nats-io/nats.go/jetstream"
|
||||
|
||||
stdio "git.novox.be/novox/mesh-sdk/go"
|
||||
"git.novox.be/novox/mesh-sdk/go/operatorask"
|
||||
)
|
||||
|
||||
// The teller's three bus adapters for mesh-sdk go/asker, which in a bundle reaches the bus through the runtime: here the
|
||||
@@ -20,9 +20,6 @@ import (
|
||||
// the seat operator-channel (mesh-control internal/broker seattraffic.go SeatTrafficOf): publish its own ask, cancel and
|
||||
// condition, hear its own decided event, and read the router's record of its own asks.
|
||||
|
||||
// callTimeout bounds a read of the router's record.
|
||||
const callTimeout = 30 * time.Second
|
||||
|
||||
// publishVia publishes an accept of the seat, de-duplicated by id: the same id twice is one message.
|
||||
func publishVia(ctx context.Context, js jetstream.JetStream) func(subject string, body any, id string) error {
|
||||
return func(subject string, body any, id string) error {
|
||||
@@ -39,30 +36,20 @@ func publishVia(ctx context.Context, js jetstream.JetStream) func(subject string
|
||||
}
|
||||
}
|
||||
|
||||
// messengerAsks is the router's records bucket as the seat operator-channel's manifest declares it (its `records`), the
|
||||
// one the controller composes this account's direct get on (mesh-sdk buscompose SeatTrafficOf): never `asks`, a bucket
|
||||
// no grant names (issue 774). The SDK names no constant for it.
|
||||
const messengerAsks = "messenger_asks"
|
||||
|
||||
// recordVia reads the router's record of one of this module's asks: the key `<module>.<id>` of the router's bucket
|
||||
// messenger_asks, by a direct get on the subject the controller grants for it,
|
||||
// `$JS.API.DIRECT.GET.KV_messenger_asks.$KV.messenger_asks.<module>.>`. The record is {state, warrant}. An ask the
|
||||
// router holds no record of is nil.
|
||||
func recordVia(ctx context.Context, conn *nats.Conn, module string) func(id string) (*stdio.StateEntry, error) {
|
||||
// recordVia reads the router's record of one of this module's asks under the module's own name: the key `<module>.<id>` of the
|
||||
// state asks of the seat operator-channel, in the bucket of the seat's holder on the seat list (mesh-sdk's operatorask, design
|
||||
// 47 G A-101, A-74); no bucket's name is in the program. The record is {state, warrant}; an ask the router holds no record
|
||||
// of is nil. call asks the seat list's verb (process.Conn.Call) and ask is the get of the record (process.Conn.Ask), which the
|
||||
// controller grants for a module that uses the seat.
|
||||
func recordVia(call operatorask.Caller, ask func(bucket, id string) (*stdio.StateEntry, error)) func(id string) (*stdio.StateEntry, error) {
|
||||
contract := operatorask.Contract{Lists: "seat:" + mma.Seat + "." + mma.VerbAssignments, ListBody: mma.AssignmentsInput{Seats: mma.AssignmentsInputSeatsTrue},
|
||||
Seat: oc.Seat, StateAsks: oc.StateAsks}
|
||||
return func(id string) (*stdio.StateEntry, error) {
|
||||
ctx, cancel := context.WithTimeout(ctx, callTimeout)
|
||||
defer cancel()
|
||||
msg, err := conn.RequestWithContext(ctx, "$JS.API.DIRECT.GET.KV_"+messengerAsks+".$KV."+messengerAsks+"."+module+"."+id, nil)
|
||||
if errors.Is(err, nats.ErrNoResponders) {
|
||||
return nil, nil
|
||||
}
|
||||
bucket, err := operatorask.Bucket(call, contract)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
if status := msg.Header.Get("Status"); status == "404" || len(msg.Data) == 0 {
|
||||
return nil, nil
|
||||
}
|
||||
return &stdio.StateEntry{Key: id, Value: msg.Data}, nil
|
||||
return ask(bucket, id)
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
@@ -8,6 +8,7 @@ import (
|
||||
"fmt"
|
||||
occ "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"
|
||||
stdio "git.novox.be/novox/mesh-sdk/go"
|
||||
"git.novox.be/novox/mesh-sdk/go/views/viewstest"
|
||||
"os"
|
||||
"path/filepath"
|
||||
@@ -178,13 +179,16 @@ const (
|
||||
seatCancel = "mesh.seat.operator-channel.accept.cancel.mesh-delivery-teller"
|
||||
seatCondition = "mesh.seat.operator-channel.accept.condition.mesh-delivery-teller"
|
||||
seatDecided = "mesh.seat.operator-channel.event.decided.mesh-delivery-teller"
|
||||
// routerAsksBucket is the router's state asks in the bucket of the seat operator-channel's holder on the seat list (here ops-router):
|
||||
// the holder's module name and the seat's state, never a name in the program (design 47 G A-101, A-74).
|
||||
routerAsksBucket = "ops-router_asks"
|
||||
)
|
||||
|
||||
func fullConf(dir string) string {
|
||||
// ends is the teller's own state (manifest state entry ends), which the controller makes before it starts: a bed
|
||||
// without it held the start ten seconds on a read the bus never answered, and the tests raced that stall.
|
||||
teller := append(append(append(append(keepsOf(tellerAnswers), keepsOf(tellerAsked)...), keepsOf(ends.Bucket)...), readOf(gateAsksKV)...),
|
||||
seatAsk, seatCancel, seatCondition, "$JS.API.DIRECT.GET.KV_"+messengerAsks+".$KV."+messengerAsks+".mesh-delivery-teller.>")
|
||||
seatAsk, seatCancel, seatCondition, "$JS.API.DIRECT.GET.KV_"+routerAsksBucket+".$KV."+routerAsksBucket+".mesh-delivery-teller.>")
|
||||
// The durable consumer on EVENTS the controller makes for the seat it uses, pulled and acknowledged (issue 1126).
|
||||
teller = append(teller, "$JS.API.CONSUMER.INFO.EVENTS."+tellerConsumer, "$JS.API.CONSUMER.MSG.NEXT.EVENTS."+tellerConsumer, "$JS.ACK.EVENTS."+tellerConsumer+".>")
|
||||
// The finding writer finds the stream of the keeper's findings-asked (design 47 G §384): the bed has none, so it is
|
||||
@@ -261,7 +265,7 @@ func startFull(t *testing.T) *fullBed {
|
||||
}
|
||||
ctx, cancel := context.WithTimeout(context.Background(), 60*time.Second)
|
||||
t.Cleanup(cancel)
|
||||
for name, history := range map[string]uint8{gateAsksKV: gateAsksHistory, tellerAnswers: answersHistory, tellerAsked: 1, ends.Bucket: 1, messengerAsks: 1, assignerAsks: assignerAsksHistory, viewAnswers: viewAnswersHistory, firstViewsBucket: 1, findingsBucket: 1} {
|
||||
for name, history := range map[string]uint8{gateAsksKV: gateAsksHistory, tellerAnswers: answersHistory, tellerAsked: 1, ends.Bucket: 1, routerAsksBucket: 1, assignerAsks: assignerAsksHistory, viewAnswers: viewAnswersHistory, firstViewsBucket: 1, findingsBucket: 1} {
|
||||
if _, err := js.CreateKeyValue(ctx, jetstream.KeyValueConfig{Bucket: name, History: history}); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
@@ -376,7 +380,7 @@ func TestTellerCarriesAnAskAndWritesTheAnswer(t *testing.T) { // 1343/AC1, AC2
|
||||
t.Cleanup(func() { rn.finish(t, 10*time.Second) })
|
||||
|
||||
// Issue 774: the router's record is read on the teller's own credential from the bucket the controller grants it.
|
||||
if mkv, err := b.js.KeyValue(b.ctx, messengerAsks); err != nil {
|
||||
if mkv, err := b.js.KeyValue(b.ctx, routerAsksBucket); err != nil {
|
||||
t.Fatal(err)
|
||||
} else if _, err := mkv.Put(b.ctx, "mesh-delivery-teller.r1", []byte(`{"state":"open"}`)); err != nil {
|
||||
t.Fatal(err)
|
||||
@@ -386,7 +390,20 @@ func TestTellerCarriesAnAskAndWritesTheAnswer(t *testing.T) { // 1343/AC1, AC2
|
||||
t.Fatal(err)
|
||||
}
|
||||
t.Cleanup(tc.Close)
|
||||
if rec, err := recordVia(b.ctx, tc, "mesh-delivery-teller")("r1"); err != nil || rec == nil {
|
||||
seatList := func(key string, _ any) (json.RawMessage, error) {
|
||||
if key != "seat:mesh-module-assigner.assignments" {
|
||||
t.Errorf("the router's holder was asked of %s, want the seat mesh-module-assigner's assignments", key)
|
||||
}
|
||||
return json.RawMessage(`{"seats":[{"seat":"operator-channel","scope":"mesh","holders":[{"module":"ops-router","node":"n2"}]}]}`), nil
|
||||
}
|
||||
getRecord := func(bucket, id string) (*stdio.StateEntry, error) {
|
||||
msg, err := tc.RequestWithContext(b.ctx, "$JS.API.DIRECT.GET.KV_"+bucket+".$KV."+bucket+".mesh-delivery-teller."+id, nil)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
return &stdio.StateEntry{Key: id, Value: msg.Data}, nil
|
||||
}
|
||||
if rec, err := recordVia(seatList, getRecord)("r1"); err != nil || rec == nil {
|
||||
t.Fatalf("the router's record read as the teller: %v, %v; want it", rec, err)
|
||||
}
|
||||
|
||||
|
||||
@@ -76,8 +76,15 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro
|
||||
// started once both are registered.
|
||||
hub := process.NewEventHub(logger.Printf)
|
||||
defer hub.Stop()
|
||||
// The assigner's asks are its state, read by the first view of its read tool and then its views family, never its
|
||||
// bucket (ADR 0615 §1, DLV-M3): the picture reads as the bucket did.
|
||||
calls, err := process.Connect(cred, nats.Name(cred.Node+"."+cred.Module+".calls"), nats.RetryOnFailedConnect(true))
|
||||
if err != nil {
|
||||
return fmt.Errorf("the teller's connection for its tool calls: %w", err)
|
||||
}
|
||||
defer calls.Close()
|
||||
client := &occ.Client{Module: tellerModule, State: state, Publish: publish,
|
||||
Record: recordVia(ctx, conn, tellerModule), Subscribe: decidedVia(hub.Handle, logger.Printf)}
|
||||
Record: recordVia(calls.Call, calls.Ask), Subscribe: decidedVia(hub.Handle, logger.Printf)}
|
||||
t := &Teller{Asker: client, Store: state, Answers: kvAnswers{ctx: ctx, kv: kvs[0]}, Gate: kvGate{ctx: ctx, kv: gate},
|
||||
Say: sayVia(ctx, js, tellerModule), Logf: logger.Printf, Publish: publish,
|
||||
// The teller's own events (its manifest's emits), on EVENTS under its module's subject.
|
||||
@@ -92,13 +99,6 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro
|
||||
// The view's answers to the assigner's asks (issue 1297 B10): both buckets are the controller's to create, and a bucket not
|
||||
// there is said and leaves the teller without that path rather than without the gate's asks (a wait for it needs a timer).
|
||||
failed := make(chan error, 4)
|
||||
// The assigner's asks are its state, read by the first view of its read tool and then its views family, never its
|
||||
// bucket (ADR 0615 §1, DLV-M3): the picture reads as the bucket did.
|
||||
calls, err := process.Connect(cred, nats.Name(cred.Node+"."+cred.Module+".calls"), nats.RetryOnFailedConnect(true))
|
||||
if err != nil {
|
||||
return fmt.Errorf("the teller's connection for its tool calls: %w", err)
|
||||
}
|
||||
defer calls.Close()
|
||||
t.Answered = func(key, askID string, answer []byte) error {
|
||||
return calls.Emit(answeredEvent, Answered{Key: key, Ask: askID, Answer: answer}, stdio.EmitOptions{MessageID: answeredID(key, askID)})
|
||||
}
|
||||
|
||||
@@ -4,6 +4,7 @@ import (
|
||||
"encoding/json"
|
||||
"os"
|
||||
"path/filepath"
|
||||
"reflect"
|
||||
"slices"
|
||||
"strings"
|
||||
"testing"
|
||||
@@ -82,11 +83,17 @@ func theManifestSaysWhatTheCodeDoes(t *testing.T) {
|
||||
Scope string `json:"scope"`
|
||||
Serves []string `json:"serves"`
|
||||
} `json:"claims"`
|
||||
Emits []string `json:"emits"`
|
||||
Reads []string `json:"reads"`
|
||||
Invokes []string `json:"invokes"`
|
||||
Tools []string `json:"tools"`
|
||||
State []json.RawMessage `json:"state"`
|
||||
Emits []string `json:"emits"`
|
||||
Reads []string `json:"reads"`
|
||||
Invokes []string `json:"invokes"`
|
||||
Tools []string `json:"tools"`
|
||||
State []json.RawMessage `json:"state"`
|
||||
Requires []struct {
|
||||
Name string `json:"name"`
|
||||
Storage string `json:"storage"`
|
||||
Template string `json:"template"`
|
||||
Options map[string]any `json:"options"`
|
||||
} `json:"requires"`
|
||||
}
|
||||
if err := json.Unmarshal(raw, &m); err != nil {
|
||||
t.Fatal(err)
|
||||
@@ -132,7 +139,7 @@ func theManifestSaysWhatTheCodeDoes(t *testing.T) {
|
||||
t.Errorf("the verb %s has no tool", v)
|
||||
}
|
||||
}
|
||||
if len(m.State) != 3 || string(m.State[0]) == "" || !slices.ContainsFunc(m.State, func(s json.RawMessage) bool {
|
||||
if len(m.State) != 2 || string(m.State[0]) == "" || !slices.ContainsFunc(m.State, func(s json.RawMessage) bool {
|
||||
var d struct {
|
||||
Name string `json:"name"`
|
||||
TTL bool `json:"key-ttl"`
|
||||
@@ -141,4 +148,12 @@ func theManifestSaysWhatTheCodeDoes(t *testing.T) {
|
||||
}) {
|
||||
t.Errorf("state %s: enrolments must allow a time to live per key", m.State)
|
||||
}
|
||||
// Design 47 G §383, §387 (A-30): its findings are the requires findings (state) and finding-events (the three finding
|
||||
// events, 7 days, 16 MiB), not a state or a stream of its own.
|
||||
if len(m.Requires) != 2 || m.Requires[0].Name != "findings" || m.Requires[0].Storage != "key-value" || m.Requires[0].Template != "state" ||
|
||||
m.Requires[1].Name != "finding-events" || m.Requires[1].Storage != "stream" || m.Requires[1].Template != "events" ||
|
||||
m.Requires[1].Options["max-age-days"] != float64(7) || m.Requires[1].Options["max-mib"] != float64(16) ||
|
||||
!reflect.DeepEqual(m.Requires[1].Options["events"], []any{"finding-written.*", "finding-cleared.*", "findings-stated.*"}) {
|
||||
t.Errorf("requires %+v: want the findings state and the finding stream", m.Requires)
|
||||
}
|
||||
}
|
||||
|
||||
@@ -85,6 +85,17 @@ func engineGrants(p Principal) (pub, sub []string) {
|
||||
sub = append(sub, k.Filter)
|
||||
}
|
||||
}
|
||||
// And the kept events of other seats it follows (the recovery's ended and restored): each stream read through an ordered
|
||||
// consumer made for the read, with no durable of its own, and the stream's own info (design 47 G §554, A-108).
|
||||
for _, h := range p.EngineHears {
|
||||
if !safeSubject.MatchString(h.Stream) {
|
||||
continue
|
||||
}
|
||||
pub = append(pub, "$JS.API.STREAM.INFO."+h.Stream,
|
||||
"$JS.API.CONSUMER.CREATE."+h.Stream, "$JS.API.CONSUMER.CREATE."+h.Stream+".>",
|
||||
"$JS.API.CONSUMER.MSG.NEXT."+h.Stream+".*", "$JS.API.CONSUMER.INFO."+h.Stream+".*",
|
||||
"$JS.API.CONSUMER.DELETE."+h.Stream+".*")
|
||||
}
|
||||
sub = append(sub, announcing(EngineSeat)...)
|
||||
pub = append(pub, discovering()...)
|
||||
tp, ts := seatTrafficGrants(SeatTrafficOf(occ.EngineAsker(p.Node), nil, p.Uses, nil))
|
||||
|
||||
@@ -4,6 +4,8 @@ import (
|
||||
"fmt"
|
||||
"sort"
|
||||
"strings"
|
||||
|
||||
"git.novox.be/novox/mesh-sdk/go/seatdef"
|
||||
)
|
||||
|
||||
// A module's own stream on the bus (novox/hq ADR 0363).
|
||||
@@ -260,6 +262,11 @@ func heardGrants(p Principal, hears []Heard) []string {
|
||||
"$JS.API.CONSUMER.INFO."+h.Stream+"."+durable,
|
||||
"$JS.API.CONSUMER.MSG.NEXT."+h.Stream+"."+durable,
|
||||
"$JS.ACK."+h.Stream+"."+durable+".>")
|
||||
if strings.HasPrefix(h.Stream, seatdef.KeptStreamPrefix) {
|
||||
// A seat's kept event is heard through process.HearSeatEvents, which reads the stream's own info to see whether it is
|
||||
// kept last per subject (design 47 G §641, A-19): the info of this stream alone.
|
||||
out = append(out, "$JS.API.STREAM.INFO."+h.Stream)
|
||||
}
|
||||
if h.LastPerSubject {
|
||||
// A stream kept last per subject is read whole at each start through an ordered consumer, made for that one read
|
||||
// and with no durable name: its creation, its next message, its info and the deletion of its own (design 47 G §556, A-4).
|
||||
|
||||
@@ -124,6 +124,9 @@ type Principal struct {
|
||||
// (engineGrants).
|
||||
EngineEvents map[string]string
|
||||
EngineKept []KeptEvent
|
||||
// EngineHears are the kept events of other seats its engine consumes (design 47 G §554, A-108: the events ended and
|
||||
// restored of the seat mesh-recovery), each stream followed through an ordered consumer made for the read.
|
||||
EngineHears []Heard
|
||||
// Ledgers are the local names of the ledgers its module asks for in `requires`, for a KindLedger principal alone
|
||||
// (issue 1966 D5): its whole authority is publishing on those ledgers' histories and snapshots (LedgerPublishGrants).
|
||||
Ledgers []string
|
||||
|
||||
@@ -21,3 +21,17 @@ func TestAReaderOfAStreamKeptLastPerSubjectIsGrantedItsOrderedConsumer(t *testin
|
||||
t.Errorf("a reader of another stream is granted an ordered consumer: %v", plain)
|
||||
}
|
||||
}
|
||||
|
||||
// A reader of a seat's kept event is granted the info of that stream, which process.HearSeatEvents reads to see whether the
|
||||
// stream is kept last per subject, and no stream's info besides (design 47 G §641, A-19).
|
||||
func TestAReaderOfASeatsKeptEventIsGrantedItsStreamInfo(t *testing.T) {
|
||||
p := Principal{Module: "m", Node: "a"}
|
||||
kept := heardGrants(p, []Heard{{Stream: "SEAT_mesh-delivery_owed", Filters: []string{"f"}}})
|
||||
if !slices.Contains(kept, "$JS.API.STREAM.INFO.SEAT_mesh-delivery_owed") {
|
||||
t.Errorf("a reader of a seat's kept event is not granted its stream's info: %v", kept)
|
||||
}
|
||||
other := heardGrants(p, []Heard{{Stream: "mesh-x-events", Filters: []string{"f"}}})
|
||||
if slices.Contains(other, "$JS.API.STREAM.INFO.mesh-x-events") {
|
||||
t.Errorf("a reader of a module's stream is granted its info: %v", other)
|
||||
}
|
||||
}
|
||||
|
||||
@@ -0,0 +1,49 @@
|
||||
package buscompose
|
||||
|
||||
import (
|
||||
"testing"
|
||||
)
|
||||
|
||||
// Design 47 G §406, §554, A-108: the node-engine of a node is granted to hear the seat mesh-recovery's kept events ended and
|
||||
// restored through an ordered consumer, to ask the verbs snapshot-kept, restore-kept and kept-points of the seat, and to call
|
||||
// the catch-up tool of a module placed on its node, on its own node's tool subject and no other module's tool.
|
||||
func TestTheNodeEngineIsGrantedWhatItNeedsOfTheRecoverySeat(t *testing.T) {
|
||||
kept := []KeptEvent{
|
||||
{Seat: "mesh-recovery", Event: "ended", Stream: "SEAT_mesh-recovery_ended", Filter: "mesh.seat.mesh-recovery.event.ended.*.*"},
|
||||
{Seat: "mesh-recovery", Event: "restored", Stream: "SEAT_mesh-recovery_restored", Filter: "mesh.seat.mesh-recovery.event.restored.*.*"},
|
||||
}
|
||||
records := Records{Nodes: []string{"one"}, Kept: kept,
|
||||
Engine: &Declared{Module: EngineModule, Consumes: []string{"mesh-recovery.ended", "mesh-recovery.restored"},
|
||||
Invokes: []string{"seat:mesh-recovery.snapshot-kept", "seat:mesh-recovery.restore-kept", "seat:mesh-recovery.kept-points"}},
|
||||
Assigned: map[string][]Declared{"one": {{Module: "keeper", CatchUps: []string{"catch_up"}}, {Module: "plain"}}}}
|
||||
users, err := Users(records)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
var node Principal
|
||||
for _, u := range users {
|
||||
if u.Kind == KindNode {
|
||||
node = u
|
||||
}
|
||||
}
|
||||
perms, err := PermissionsFor(node)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
for _, want := range []string{
|
||||
"$JS.API.CONSUMER.CREATE.SEAT_mesh-recovery_ended", "$JS.API.CONSUMER.CREATE.SEAT_mesh-recovery_ended.>",
|
||||
"$JS.API.CONSUMER.MSG.NEXT.SEAT_mesh-recovery_ended.*", "$JS.API.CONSUMER.DELETE.SEAT_mesh-recovery_restored.*",
|
||||
"$JS.API.STREAM.INFO.SEAT_mesh-recovery_restored",
|
||||
"mesh.seat.mesh-recovery.tool.snapshot-kept", "mesh.seat.mesh-recovery.tool.restore-kept", "mesh.seat.mesh-recovery.tool.kept-points",
|
||||
"mesh.mod.keeper.tool.catch_up.one",
|
||||
} {
|
||||
if !MayPublish(perms, want) {
|
||||
t.Errorf("the node-engine of one may not publish %s", want)
|
||||
}
|
||||
}
|
||||
for _, bad := range []string{"mesh.mod.plain.tool.catch_up.one", "mesh.mod.keeper.tool.other.one", "$JS.API.CONSUMER.CREATE.SEAT_mesh-recovery_set-aside"} {
|
||||
if MayPublish(perms, bad) {
|
||||
t.Errorf("the node-engine of one may publish %s", bad)
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -53,6 +53,9 @@ type Declared struct {
|
||||
Watches []Seat
|
||||
// Invokes are the tools it calls, `<module>.<tool>` or `*` (novox/hq ADR 0152).
|
||||
Invokes []string
|
||||
// CatchUps are the module's own tools that bring a kept data forward after a restore of it, as its manifest's kept data
|
||||
// name them (design 47 G §406, §631, A-108): the node-engine on a node that places the module is granted to call each.
|
||||
CatchUps []string
|
||||
// Waits are the event families a session may wait on through the tool runner, the tool runner's module's
|
||||
// alone (novox/hq ADR 0367 §2): its account subscribes them for a call, through no consumer.
|
||||
Waits []Wait
|
||||
@@ -223,9 +226,20 @@ func Users(r Records) ([]Principal, error) {
|
||||
if r.Engine != nil {
|
||||
engineHears, engineWatches, engineInvokes = r.Engine.Consumes, r.Engine.Watches, r.Engine.Invokes
|
||||
}
|
||||
// The node-engine's hearing of kept events of other seats (the recovery's ended and restored) and its call of each
|
||||
// catch-up tool of a module placed on its node (design 47 G §406, §554, A-108), granted as the seat's grants name them.
|
||||
_, engineKeptHears := SplitKept(r.Kept, engineHears)
|
||||
engineInvokes = slices.Clone(engineInvokes)
|
||||
for _, d := range r.Assigned[node] {
|
||||
for _, tool := range d.CatchUps {
|
||||
if t := d.Module + "." + tool; !slices.Contains(engineInvokes, t) {
|
||||
engineInvokes = append(engineInvokes, t)
|
||||
}
|
||||
}
|
||||
}
|
||||
for _, login := range nodeLoginsOf(r, node) {
|
||||
out = append(out, Principal{Kind: KindNode, Node: node, Login: login, Checks: checks, Runner: runner, Consumes: engineHears,
|
||||
Watches: engineWatches, Invokes: engineInvokes, EngineEvents: r.EngineEvents, EngineKept: keptOfSeat(r.Kept, EngineSeat),
|
||||
EngineHears: engineKeptHears, Watches: engineWatches, Invokes: engineInvokes, EngineEvents: r.EngineEvents, EngineKept: keptOfSeat(r.Kept, EngineSeat),
|
||||
MovesIntoTheVault: servesMoveIn(r), StatesItsKeyToTheVault: servesVerb(r, VaultNodeKey),
|
||||
AsksItsOwnSecrets: servesVerb(r, VaultOwn), AsksTheVaultToSign: servesVerb(r, VaultSign), Uses: asksSeatOf(r),
|
||||
FollowsGrantsUserList: bus, ServedReads: servedOn(served, node), RotatesItsLogin: servesVerb(r, VaultRotateNode)})
|
||||
|
||||
@@ -0,0 +1,21 @@
|
||||
package grants
|
||||
|
||||
import (
|
||||
"slices"
|
||||
"testing"
|
||||
|
||||
"git.novox.be/novox/mesh-sdk/go/catalogue"
|
||||
)
|
||||
|
||||
// Design 47 G §406, §631, A-108: the catch-up tools a manifest's kept data name are the module's own, each once, whichever
|
||||
// kind of kept data names them.
|
||||
func TestCatchUpsAreTheToolsTheManifestsKeptDataName(t *testing.T) {
|
||||
m, err := catalogue.Decode([]byte(`{"module":"keeper","version":"1.0","state":[{"name":"seen","form":2,"catch-up":{"tool":"seen_again","arguments":{"all":true}}},"plain"],` +
|
||||
`"logs":[{"name":"lines","form":1,"catch-up":"seen_again"}],"streams":[{"name":"ev","events":["a"],"max-age-days":1,"max-mib":1,"form":1,"catch-up":{"tool":"ev_again"}}]}`))
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if got := catchUpsOf(m); !slices.Equal(got, []string{"seen_again", "ev_again"}) && !slices.Equal(got, []string{"ev_again", "seen_again"}) {
|
||||
t.Errorf("catchUpsOf = %v, want seen_again and ev_again once each", got)
|
||||
}
|
||||
}
|
||||
@@ -156,6 +156,7 @@ func declaredFor(m catalogue.Manifest, seats map[string]catalogue.SeatDeclaratio
|
||||
Ledgers: ledgersOf(m),
|
||||
}
|
||||
d.Loads = loadsABundle(m)
|
||||
d.CatchUps = catchUpsOf(m)
|
||||
if _, reads := m.OwnSecrets["broker"]; !reads {
|
||||
d.NoAccount = true
|
||||
}
|
||||
@@ -608,3 +609,15 @@ func unionFields(dst, src reflect.Value) {
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// catchUpsOf are the catch-up tools a manifest's kept data name, each once (design 47 G §406, §631): the module's own tools
|
||||
// the node-engine calls after a restore of that data, which it is granted to call on its own node (A-108).
|
||||
func catchUpsOf(m catalogue.Manifest) []string {
|
||||
var tools []string
|
||||
for _, k := range catalogue.KeptDeclarations(m) {
|
||||
if k.CatchUp != nil && k.CatchUp.Tool != "" && !slices.Contains(tools, k.CatchUp.Tool) {
|
||||
tools = append(tools, k.CatchUp.Tool)
|
||||
}
|
||||
}
|
||||
return tools
|
||||
}
|
||||
|
||||
@@ -0,0 +1,32 @@
|
||||
package grants
|
||||
|
||||
import (
|
||||
"os"
|
||||
"path/filepath"
|
||||
"testing"
|
||||
|
||||
"git.novox.be/novox/mesh-sdk/go/seatdef"
|
||||
)
|
||||
|
||||
// Design 47 G A-114 (§693, A-61): the seat mesh-wants' `wanted` stream keeps the last message of each subject with no age
|
||||
// limit, so each want's last word is read whole at a start.
|
||||
func TestWantedKeepsEachWantsLastWordWithNoAgeLimit(t *testing.T) {
|
||||
raw, err := os.ReadFile(filepath.Join("..", "..", "..", "..", "contracts", "seats", "mesh-wants", "seat.json"))
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
d, err := seatdef.Parse(raw)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
for _, e := range d.Events {
|
||||
if e.Name != "wanted" {
|
||||
continue
|
||||
}
|
||||
if !e.KeptLastPerSubject() || e.KeptAgeDays() != 0 {
|
||||
t.Errorf("wanted keeps last-per-subject %v for %d days, want true and 0 (no age limit)", e.KeptLastPerSubject(), e.KeptAgeDays())
|
||||
}
|
||||
return
|
||||
}
|
||||
t.Fatal("the seat mesh-wants has no event wanted")
|
||||
}
|
||||
Reference in New Issue
Block a user