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), refusing every caller while the list is unread. The assigner (mesh-wants input), the builder (mesh-builds reserved) and the indexer (mesh-catalog input) hand it the definition their seat list holds once heard; the pipeline runner, which followed none, follows one over its calls connection and hub, read again once its events are live, and its manifest invokes seat-definitions and consumes the list's two events. The runner's and the indexer's integration beds answer the seat list, and the indexer's input test calls with a caller the bus proved (A-179). 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:
@@ -289,14 +289,17 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro
|
||||
// The inputs of the composition steps it asks a build seat to run: the seat mesh-wants' verb input is answered here in chunks
|
||||
// of the core connection, of any size (design 47 G §354, §712, A-237).
|
||||
inputs := inputsOver(kvs[buckets.Inputs], js, account.Module, logger.Printf)
|
||||
// The seat list (seatlist.go), followed in follow over the connection and the hub: the verb input's callers are checked
|
||||
// against the seat's definition it holds (A-186).
|
||||
gate := seatlist.NewPart("assigner", seatListAgainCallers)
|
||||
if inputs != nil {
|
||||
stopInput, err := said.AnswerSeat(meshwants.Seat, map[string]process.SeatHandler{meshwants.VerbInput: inputs.verb(conn, buscalls.SeatAckWithin)})
|
||||
stopInput, err := said.AnswerSeat(meshwants.Seat, seatDefinition(gate, meshwants.Seat), map[string]process.SeatHandler{meshwants.VerbInput: inputs.verb(conn, buscalls.SeatAckWithin)})
|
||||
if err != nil {
|
||||
return fmt.Errorf("answering the verb input: %w", err)
|
||||
}
|
||||
defer stopInput()
|
||||
}
|
||||
return follow(ctx, js, conn, kvs, findings.Say(sayVia(ctx, js)), findings, emitting{said}, iv, inputs, logger, getenv)
|
||||
return follow(ctx, js, conn, kvs, findings.Say(sayVia(ctx, js)), findings, emitting{said}, iv, inputs, gate, logger, getenv)
|
||||
}
|
||||
|
||||
// findingsOf is where the assigner's said conditions go as findings (issue 1952 WM6): its findings bucket, which the
|
||||
@@ -379,7 +382,7 @@ var watched = []string{
|
||||
|
||||
// follow watches every bucket a pass reads and makes a pass at start, once every watch handed what stands, and after
|
||||
// each change; changes that come during a pass make one more. It returns when ctx ends or a watch ends.
|
||||
func follow(ctx context.Context, js jetstream.JetStream, nc process.Subscriber, kvs kvSet, say func(oc.Condition) error, findings assign.Findings, bus caller, iv *catalogViews, inputs *inputKeeper, logger *log.Logger, getenv func(string) string) error {
|
||||
func follow(ctx context.Context, js jetstream.JetStream, nc process.Subscriber, kvs kvSet, say func(oc.Condition) error, findings assign.Findings, bus caller, iv *catalogViews, inputs *inputKeeper, gate *seatlist.Part, logger *log.Logger, getenv func(string) string) error {
|
||||
w := newWorld()
|
||||
dirty := make(chan struct{}, 1)
|
||||
wake := func() {
|
||||
@@ -498,7 +501,6 @@ func follow(ctx context.Context, js jetstream.JetStream, nc process.Subscriber,
|
||||
hub := process.NewEventHub(logger.Printf)
|
||||
// The seat list (seatlist.go), read over the connection and heard on the hub, given to catalogue at every change, and a
|
||||
// pass made then: registered before the hub starts.
|
||||
gate := seatlist.NewPart("assigner", seatListAgainCallers)
|
||||
if bus != nil {
|
||||
gate.Follow(seatlist.ProcessBus(bus.Call, hub), wake, logger.Printf)
|
||||
}
|
||||
|
||||
@@ -28,6 +28,7 @@ import (
|
||||
oc "git.novox.be/novox/mesh-contracts/operator-and-conversation/go/seats/operatorchannel"
|
||||
"git.novox.be/novox/mesh-sdk/go/catalogue"
|
||||
"git.novox.be/novox/mesh-sdk/go/process"
|
||||
"git.novox.be/novox/mesh-sdk/go/seatlist"
|
||||
)
|
||||
|
||||
// 1297/B2a: the buckets the program opens are exactly the manifest's reads and states, as the broker names them
|
||||
@@ -124,7 +125,7 @@ func TestOpensWhatItsManifestNames(t *testing.T) {
|
||||
}
|
||||
kvs[buckets.Puts].(*watchKV).closed = true
|
||||
kvs[buckets.Wants].(*watchKV).wrote = wrote
|
||||
go follow(ctx, nil, nil, kvs, func(oc.Condition) error { return nil }, assign.Findings{}, nil, nil, nil, log.New(io.Discard, "", 0), func(string) string { return "" })
|
||||
go follow(ctx, nil, nil, kvs, func(oc.Condition) error { return nil }, assign.Findings{}, nil, nil, nil, seatlist.NewPart("assigner", seatListAgainCallers), log.New(io.Discard, "", 0), func(string) string { return "" })
|
||||
select {
|
||||
case key := <-wrote:
|
||||
if key != "want.ace.node" {
|
||||
|
||||
@@ -1,6 +1,11 @@
|
||||
package main
|
||||
|
||||
import stdio "git.novox.be/novox/mesh-sdk/go"
|
||||
import (
|
||||
"git.novox.be/novox/mesh-sdk/go/seatdef"
|
||||
"git.novox.be/novox/mesh-sdk/go/seatlist"
|
||||
|
||||
stdio "git.novox.be/novox/mesh-sdk/go"
|
||||
)
|
||||
|
||||
// The assigner reads the seat list as every part does, through mesh-sdk's one wiring (go/seatlist.Part; design 47 G §204): its
|
||||
// reader over the seat mesh-catalog's verb seat-definitions and its two events, its tools `assigner_seat_list` and
|
||||
@@ -11,3 +16,17 @@ import stdio "git.novox.be/novox/mesh-sdk/go"
|
||||
// A call through the mesh MCP server is proved only as its machine's tool runner (<node>.node-tools, issue 627), so
|
||||
// node-tools stands for an operator's session until a session is proved (issue 1966).
|
||||
var seatListAgainCallers = []string{stdio.AnyPerson, "node-tools", "mesh-healers"}
|
||||
|
||||
// seatDefinition is the definition of seat the gate's seat list holds, which the seat's verbs answered here check their
|
||||
// callers against (process.Conn.AnswerSeat; design 47 G lead answers A-179, A-186): false until the seat mesh-catalog has
|
||||
// answered the list, so the assigner answers a verb only once its seat list is read, and false for a seat it does not hold.
|
||||
func seatDefinition(gate *seatlist.Part, seat string) func() (seatdef.Definition, bool) {
|
||||
return func() (seatdef.Definition, bool) {
|
||||
r := gate.Reader()
|
||||
if r == nil || !r.Status().Heard {
|
||||
return seatdef.Definition{}, false
|
||||
}
|
||||
d, ok := r.Definitions()[seat]
|
||||
return d, ok
|
||||
}
|
||||
}
|
||||
|
||||
@@ -94,3 +94,21 @@ func TestTheSeatListIsWiredByTheSDKAlone(t *testing.T) {
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// A-179, A-186: the definition the verb input's callers are checked 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.
|
||||
func TestTheAnsweredSeatsDefinitionIsTheSeatListsOnceRead(t *testing.T) {
|
||||
if _, ok := seatDefinition(seatlist.NewPart("assigner", seatListAgainCallers), "mesh-wants")(); ok {
|
||||
t.Error("a definition is answered before the seat list's reader started")
|
||||
}
|
||||
if _, ok := seatDefinition(gateOver(t, true), "mesh-wants")(); ok {
|
||||
t.Error("a definition is answered while the seat mesh-catalog has not answered the list")
|
||||
}
|
||||
read := gateOver(t, false)
|
||||
if d, ok := seatDefinition(read, "mesh-wants")(); !ok || d.Seat != "mesh-wants" {
|
||||
t.Errorf("the read list's mesh-wants is answered as %v, %v", d.Seat, ok)
|
||||
}
|
||||
if _, ok := seatDefinition(read, "no-such-seat")(); ok {
|
||||
t.Error("a seat the list does not hold is answered a definition")
|
||||
}
|
||||
}
|
||||
|
||||
@@ -405,7 +405,7 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro
|
||||
b.StepEntry = newEntryWaits(branches, half.manifests).entry
|
||||
// The seat mesh-builds' verb reserved, answered here (A-115, A-127): the reservations, the catalog half's carriers and the seat
|
||||
// mesh-catalog's versions.
|
||||
stopReserved, err := calls.AnswerSeat(meshbuilds.Seat, map[string]process.SeatHandler{
|
||||
stopReserved, err := calls.AnswerSeat(meshbuilds.Seat, seatGate.definition(meshbuilds.Seat), map[string]process.SeatHandler{
|
||||
meshbuilds.VerbReserved: reservedVerb(ctx, reservedReader(ctx, reserved.r.Store), versions.CarriersFrom(versions.EntriesOf(branches)), catalogReleased(calls.Call))})
|
||||
if err != nil {
|
||||
return fmt.Errorf("answering the verb reserved: %w", err)
|
||||
|
||||
@@ -24,3 +24,17 @@ func (g *seatGate) definitions() map[string]seatdef.Definition {
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
// definition is the definition of seat the list holds, which the seat's verbs answered here check their callers against
|
||||
// (process.Conn.AnswerSeat; design 47 G lead answers A-179, A-186): false until the seat mesh-catalog has answered the list, so
|
||||
// the builder answers a verb only once its seat list is read, and false for a seat it does not hold.
|
||||
func (g *seatGate) definition(seat string) func() (seatdef.Definition, bool) {
|
||||
return func() (seatdef.Definition, bool) {
|
||||
r := g.Reader()
|
||||
if r == nil || !r.Status().Heard {
|
||||
return seatdef.Definition{}, false
|
||||
}
|
||||
d, ok := r.Definitions()[seat]
|
||||
return d, ok
|
||||
}
|
||||
}
|
||||
|
||||
@@ -0,0 +1,50 @@
|
||||
package main
|
||||
|
||||
import (
|
||||
"encoding/json"
|
||||
"errors"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
"git.novox.be/novox/mesh-sdk/go/busretry"
|
||||
"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 }
|
||||
|
||||
// A-179, A-186: the definition the verb reserved's callers are checked 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.
|
||||
func TestTheAnsweredSeatsDefinitionIsTheSeatListsOnceRead(t *testing.T) {
|
||||
t.Cleanup(busretry.Waits(20*time.Millisecond, time.Millisecond, 5*time.Millisecond))
|
||||
follow := func(silent bool) *seatGate {
|
||||
g := newSeatGate()
|
||||
if r := g.Follow(listingCatalog{silent: silent}, nil, t.Logf); r != nil {
|
||||
<-r.Ready()
|
||||
}
|
||||
return g
|
||||
}
|
||||
if _, ok := newSeatGate().definition(seatlist.Seat)(); ok {
|
||||
t.Error("a definition is answered before the seat list's reader started")
|
||||
}
|
||||
if _, ok := follow(true).definition(seatlist.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(seatlist.Seat)(); !ok || d.Seat != seatlist.Seat {
|
||||
t.Errorf("the read list's %s is answered as %q, %v", seatlist.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")
|
||||
}
|
||||
}
|
||||
@@ -10,6 +10,8 @@ import (
|
||||
"encoding/base64"
|
||||
"encoding/hex"
|
||||
"encoding/json"
|
||||
"errors"
|
||||
"strings"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
@@ -38,6 +40,10 @@ func TestIndexerAnswersARunsInputOverOneMiBInChunks(t *testing.T) {
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
// The indexer's own seat list (A-186), its state seat-definitions: empty, so the bootstrapped mesh-catalog stands.
|
||||
if _, err := js.CreateKeyValue(ctx, jetstream.KeyValueConfig{Bucket: index.SeatDefinitionsBucket, History: 5}); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
objs, err := js.CreateObjectStore(ctx, jetstream.ObjectStoreConfig{Bucket: catalogue.BusName(index.Module, catalogue.StateObjectsName(index.InputsState))})
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
@@ -62,12 +68,17 @@ func TestIndexerAnswersARunsInputOverOneMiBInChunks(t *testing.T) {
|
||||
Error string `json:"error"`
|
||||
}
|
||||
for deadline := time.Now().Add(30 * time.Second); ; time.Sleep(100 * time.Millisecond) {
|
||||
m, err := nc.Request("mesh.seat.mesh-catalog.tool.input", req, 5*time.Second)
|
||||
// A caller the bus proved, on the subject naming it (A-179); refused while the indexer's seat list is unread (A-186).
|
||||
m, err := nc.Request("mesh.seat.mesh-catalog.call.input.agent~dev", req, 5*time.Second)
|
||||
if err == nil {
|
||||
reply.Error = ""
|
||||
if err := json.Unmarshal(m.Data, &reply); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
break
|
||||
if !strings.Contains(reply.Error, "not yet read") {
|
||||
break
|
||||
}
|
||||
err = errors.New(reply.Error)
|
||||
}
|
||||
if time.Now().After(deadline) {
|
||||
t.Fatalf("the verb input is not answered within 30 s: %v", err)
|
||||
|
||||
@@ -450,7 +450,7 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro
|
||||
// connection, of any size (design 47 G §712, A-237).
|
||||
inputsStore := awaitedStore{b: process.AwaitBucket(ctx, js, conn, index.InputsBucket, "no input of a run ask is kept, so the verb input answers none", logger.Printf)}
|
||||
inObjects := &inputObjects{js: js, module: cred.Module}
|
||||
stopInput, err := calls.AnswerSeat(meshcatalog.Seat, map[string]process.SeatHandler{
|
||||
stopInput, err := calls.AnswerSeat(meshcatalog.Seat, seatDefinition(seatReader, meshcatalog.Seat), map[string]process.SeatHandler{
|
||||
meshcatalog.VerbInput: index.InputVerb(inputsStore, inObjects, conn, buscalls.SeatAckWithin, logger.Printf)})
|
||||
if err != nil {
|
||||
return fmt.Errorf("answering the verb input: %w", err)
|
||||
|
||||
@@ -10,6 +10,7 @@ import (
|
||||
"github.com/nats-io/nats.go/jetstream"
|
||||
|
||||
"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"
|
||||
|
||||
@@ -141,3 +142,17 @@ func readWhenBound(part *seatlist.Part, states ...bucket) {
|
||||
|
||||
// toolPrefix is the prefix of the indexer's tools, its manifest's slug.
|
||||
const toolPrefix = "indexer"
|
||||
|
||||
// seatDefinition is the definition of seat the indexer's seat list holds, which the seat's verbs answered here check their
|
||||
// callers against (process.Conn.AnswerSeat; design 47 G lead answers A-179, A-186): false until the list is read, so the
|
||||
// indexer answers a verb only once its seat list is read, and false for a seat it does not hold.
|
||||
func seatDefinition(part *seatlist.Part, seat string) func() (seatdef.Definition, bool) {
|
||||
return func() (seatdef.Definition, bool) {
|
||||
r := part.Reader()
|
||||
if r == nil || !r.Status().Heard {
|
||||
return seatdef.Definition{}, false
|
||||
}
|
||||
d, ok := r.Definitions()[seat]
|
||||
return d, ok
|
||||
}
|
||||
}
|
||||
|
||||
@@ -50,6 +50,13 @@ func TestTheSeatListIsReadAgainWhenItsStatesAreBound(t *testing.T) {
|
||||
if reader.Status().Heard {
|
||||
t.Fatal("the list was read with its states missing")
|
||||
}
|
||||
// A-179, A-186: the definition the verb input's callers are checked against is the list's once read, none before.
|
||||
if _, ok := seatDefinition(seatlist.NewPart(toolPrefix, nil), seatlist.Seat)(); ok {
|
||||
t.Error("a definition is answered before the seat list's reader started")
|
||||
}
|
||||
if _, ok := seatDefinition(part, seatlist.Seat)(); ok {
|
||||
t.Error("a definition is answered while the list is unread")
|
||||
}
|
||||
b := &awaitedFake{}
|
||||
readWhenBound(part, b)
|
||||
src.mu.Lock()
|
||||
@@ -63,5 +70,11 @@ func TestTheSeatListIsReadAgainWhenItsStatesAreBound(t *testing.T) {
|
||||
t.Fatal("the list was not read after its states were bound")
|
||||
}
|
||||
}
|
||||
if d, ok := seatDefinition(part, seatlist.Seat)(); !ok || d.Seat != seatlist.Seat {
|
||||
t.Errorf("the read list's %s is answered as %q, %v", seatlist.Seat, d.Seat, ok)
|
||||
}
|
||||
if _, ok := seatDefinition(part, "no-such-seat")(); ok {
|
||||
t.Error("a seat the list does not hold is answered a definition")
|
||||
}
|
||||
var _ seatlist.Bus = bus
|
||||
}
|
||||
|
||||
@@ -21,6 +21,7 @@ import (
|
||||
catalog "git.novox.be/novox/mesh-delivery/modules/mesh-delivery-contracts/go/seats/meshcatalog"
|
||||
seat "git.novox.be/novox/mesh-delivery/modules/mesh-delivery-contracts/go/seats/meshpipelinerunner"
|
||||
"git.novox.be/novox/mesh-sdk/go/process"
|
||||
"git.novox.be/novox/mesh-sdk/go/seatlist"
|
||||
|
||||
"mesh-delivery-pipeline/internal/pipeline"
|
||||
)
|
||||
@@ -246,11 +247,15 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro
|
||||
// An entry whose step ended and was not appended is appended at the start, in order (§606, §549).
|
||||
hearer.AppendUnlogged()
|
||||
go whileTheLogIsFull(ctx, fills, hearer, time.Minute)
|
||||
if err := hub.StartUntilBound(ctx, js, cred, nil, nil); err != nil {
|
||||
// The seat list, read over the calls connection and heard on the hub (seatlist.Part), which the seat's write verbs check
|
||||
// their callers against (A-186): registered before the hub starts, read again once its events are live.
|
||||
seats := seatlist.NewPart("runner", nil)
|
||||
seats.Follow(seatlist.ProcessBus(calls.Call, hub), nil, logger.Printf)
|
||||
if err := hub.StartUntilBound(ctx, js, cred, nil, seats.EventsLive); err != nil {
|
||||
logger.Printf("the consumer %s is not bound yet (%v): bound once the bus makes it", cred.EventConsumerName(), err)
|
||||
}
|
||||
// The seat's write verbs are answered here, by the process that holds the pipelines (A-107).
|
||||
stopVerbs, err := answerVerbs(calls, writeVerbs(ctx, hearer, starter, func(branch string) (pipeline.Half, bool) { return rv.Catalog.View(branch) }, calls, bounds))
|
||||
stopVerbs, err := answerVerbs(calls, seats, writeVerbs(ctx, hearer, starter, func(branch string) (pipeline.Half, bool) { return rv.Catalog.View(branch) }, calls, bounds))
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
@@ -9,6 +9,7 @@ import (
|
||||
"errors"
|
||||
"os"
|
||||
"path/filepath"
|
||||
"strings"
|
||||
"sync"
|
||||
"testing"
|
||||
"time"
|
||||
@@ -21,6 +22,7 @@ import (
|
||||
seat "git.novox.be/novox/mesh-delivery/modules/mesh-delivery-contracts/go/seats/meshpipelinerunner"
|
||||
"git.novox.be/novox/mesh-sdk/go/buscompose"
|
||||
"git.novox.be/novox/mesh-sdk/go/process"
|
||||
"git.novox.be/novox/mesh-sdk/go/seatlist"
|
||||
"git.novox.be/novox/mesh-sdk/go/testbus"
|
||||
|
||||
"mesh-delivery-pipeline/internal/pipeline"
|
||||
@@ -94,6 +96,28 @@ func TestTheRunnerPlacesAPipelineAsksItsBuildTakesItsEndAndAnswersAReview(t *tes
|
||||
answer("mesh.seat.mesh-catalog.call.registrations.*", catalog.RegistrationsAnswer{Registrations: []catalog.Registration{reg}})
|
||||
answer("mesh.seat.mesh-catalog.call.revision.*", catalog.RevisionAnswer{NoneYet: true})
|
||||
answer("mesh.seat.mesh-hand-acts.call.hand-act.*", map[string]any{})
|
||||
// The seat list (A-186): the seat mesh-catalog's seat-definitions names the runner's seat, whose definition its write
|
||||
// verbs' callers are checked against, as the contracts' file holds it.
|
||||
def, err := os.ReadFile("../../../mesh-delivery-contracts/seats/mesh-pipeline-runner/seat.json")
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if _, err := nc.Subscribe("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:runner"}
|
||||
var result any = seatlist.Listing{Revision: 1, Seats: []seatlist.Entry{entry}}
|
||||
if in.Seat != "" {
|
||||
entry.Definition = def
|
||||
result = seatlist.Answer{Seat: in.Seat, Found: in.Seat == seat.Seat, Entry: &entry, Revision: 1}
|
||||
}
|
||||
raw, _ := json.Marshal(map[string]any{"result": result})
|
||||
_ = m.Respond(raw)
|
||||
}); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
|
||||
dir := t.TempDir()
|
||||
broker := filepath.Join(dir, "broker")
|
||||
@@ -145,7 +169,9 @@ func TestTheRunnerPlacesAPipelineAsksItsBuildTakesItsEndAndAnswersAReview(t *tes
|
||||
}
|
||||
waitFor(t, ctx, logs, func() bool {
|
||||
m, err := nc.Request("mesh.seat.mesh-pipeline-runner.call.review.agent~dev", in, 2*time.Second)
|
||||
return err == nil && json.Unmarshal(m.Data, &reply) == nil
|
||||
reply.Error = ""
|
||||
// Until the runner's seat list is read, every caller is refused (A-186): asked again until it is.
|
||||
return err == nil && json.Unmarshal(m.Data, &reply) == nil && !strings.Contains(reply.Error, "not yet read")
|
||||
})
|
||||
if reply.Error != "" || reply.Result.Pipeline != key || reply.Result.Result.By != "agent.dev" || reply.Result.Result.Outcome != "passed" {
|
||||
t.Fatalf("review answered %+v (%s)", reply.Result, reply.Error)
|
||||
|
||||
@@ -12,6 +12,8 @@ import (
|
||||
catalog "git.novox.be/novox/mesh-delivery/modules/mesh-delivery-contracts/go/seats/meshcatalog"
|
||||
seat "git.novox.be/novox/mesh-delivery/modules/mesh-delivery-contracts/go/seats/meshpipelinerunner"
|
||||
"git.novox.be/novox/mesh-sdk/go/process"
|
||||
"git.novox.be/novox/mesh-sdk/go/seatdef"
|
||||
"git.novox.be/novox/mesh-sdk/go/seatlist"
|
||||
|
||||
"mesh-delivery-pipeline/internal/pipeline"
|
||||
)
|
||||
@@ -34,9 +36,23 @@ func verbOf[In any](do func(caller string, in In) (any, error)) process.SeatHand
|
||||
}
|
||||
|
||||
// answerVerbs answers the seat's write verbs over the bus, each writing the state pipelines by compare-and-set (design 47
|
||||
// G §22, §70, §306, §482, A-107). The returned stop ends the answers.
|
||||
func answerVerbs(conn *process.Conn, v pipeline.Verbs) (func(), error) {
|
||||
return conn.AnswerSeat(seat.Seat, writeHandlers(v))
|
||||
// G §22, §70, §306, §482, A-107), each call's caller checked against the seat's definition the runner's seat list holds
|
||||
// (A-179, A-186). The returned stop ends the answers.
|
||||
func answerVerbs(conn *process.Conn, seats *seatlist.Part, v pipeline.Verbs) (func(), error) {
|
||||
return conn.AnswerSeat(seat.Seat, seatDefinition(seats, seat.Seat), writeHandlers(v))
|
||||
}
|
||||
|
||||
// seatDefinition is the definition of seat the runner's seat list holds (A-186): false until the seat mesh-catalog has
|
||||
// answered the list, so the runner answers a verb only once its seat list is read, and false for a seat it does not hold.
|
||||
func seatDefinition(seats *seatlist.Part, name 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()[name]
|
||||
return d, ok
|
||||
}
|
||||
}
|
||||
|
||||
// writeHandlers are the seat's write verbs the process answers: run, review, rerun, accept, cancel, start, ask-again and
|
||||
|
||||
@@ -2,8 +2,13 @@ package main
|
||||
|
||||
import (
|
||||
"encoding/json"
|
||||
"errors"
|
||||
"slices"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
"git.novox.be/novox/mesh-sdk/go/busretry"
|
||||
"git.novox.be/novox/mesh-sdk/go/seatlist"
|
||||
|
||||
seat "git.novox.be/novox/mesh-delivery/modules/mesh-delivery-contracts/go/seats/meshpipelinerunner"
|
||||
|
||||
@@ -40,3 +45,42 @@ func TestAVerbIsAnsweredOnlyForAProvedCallerAndItsOwnInput(t *testing.T) {
|
||||
t.Errorf("answer = %v, %v, input %+v", out, err, got)
|
||||
}
|
||||
}
|
||||
|
||||
// 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 write verbs' callers are checked 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.
|
||||
func TestTheAnsweredSeatsDefinitionIsTheSeatListsOnceRead(t *testing.T) {
|
||||
t.Cleanup(busretry.Waits(20*time.Millisecond, time.Millisecond, 5*time.Millisecond))
|
||||
follow := func(silent bool) *seatlist.Part {
|
||||
p := seatlist.NewPart("runner", nil)
|
||||
if r := p.Follow(listingCatalog{silent: silent}, nil, t.Logf); r != nil {
|
||||
<-r.Ready()
|
||||
}
|
||||
return p
|
||||
}
|
||||
if _, ok := seatDefinition(seatlist.NewPart("runner", nil), seatlist.Seat)(); ok {
|
||||
t.Error("a definition is answered before the seat list's reader started")
|
||||
}
|
||||
if _, ok := seatDefinition(follow(true), seatlist.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, seatlist.Seat)(); !ok || d.Seat != seatlist.Seat {
|
||||
t.Errorf("the read list's %s is answered as %q, %v", seatlist.Seat, d.Seat, ok)
|
||||
}
|
||||
if _, ok := seatDefinition(read, seat.Seat)(); ok {
|
||||
t.Error("a seat the list does not hold is answered a definition")
|
||||
}
|
||||
}
|
||||
|
||||
@@ -78,10 +78,14 @@ func TestManifest(t *testing.T) {
|
||||
"mesh-delivery-indexer.take-in.taken", "mesh-builds.package-published", "mesh-builds.toolchain-agreed", "mesh-builds.step-ended", "mesh-delivery-builder.build.ended.**",
|
||||
"operator-channel.decided", "node-build-go.step-ended", "node-build-node.step-ended",
|
||||
"node-build-image.step-ended", "node-build-go.ask-taken", "node-build-node.ask-taken", "node-build-image.ask-taken", "node-engine.running", "mesh-wants.step-ended", "mesh-catalog.registration-changed",
|
||||
"mesh-catalog.catalog-revision", "mesh-conditions.findings-asked", "mesh-delivery.recovery-merged"})},
|
||||
"mesh-catalog.catalog-revision", "mesh-conditions.findings-asked", "mesh-delivery.recovery-merged",
|
||||
// A-186: the seat list its write verbs' callers are checked against, heard as it changes.
|
||||
"mesh-catalog.seat-definition-changed", "mesh-catalog.seat-definition-removed"})},
|
||||
// RN2 (§21, §195, §200, §77): the registrations and the catalog half asked by their verbs, the first views read and
|
||||
// asked again by hand (a hand act).
|
||||
{"invokes", equalSets(m.Invokes, []string{"seat:mesh-catalog.registrations", "seat:mesh-catalog.revision", "seat:mesh-hand-acts.hand-act",
|
||||
{"invokes", equalSets(m.Invokes, []string{"seat:mesh-catalog.registrations",
|
||||
// A-186: the seat list, read by its verb.
|
||||
"seat:mesh-catalog.seat-definitions", "seat:mesh-catalog.revision", "seat:mesh-hand-acts.hand-act",
|
||||
// RN3 (§53, §334): a publication's dependents behind.
|
||||
"seat:mesh-catalog.outdated",
|
||||
// RN10 (§71, §441, A-207): the permissions a change widens, asked of the grants part.
|
||||
|
||||
@@ -50,9 +50,12 @@
|
||||
"mesh-catalog.registration-changed",
|
||||
"mesh-catalog.catalog-revision",
|
||||
"mesh-conditions.findings-asked",
|
||||
"mesh-delivery.recovery-merged"
|
||||
"mesh-delivery.recovery-merged",
|
||||
"mesh-catalog.seat-definition-changed",
|
||||
"mesh-catalog.seat-definition-removed"
|
||||
],
|
||||
"invokes": [
|
||||
"seat:mesh-catalog.seat-definitions",
|
||||
"seat:mesh-catalog.registrations",
|
||||
"seat:mesh-catalog.revision",
|
||||
"seat:mesh-hand-acts.hand-act",
|
||||
|
||||
Reference in New Issue
Block a user