diff --git a/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/main.go b/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/main.go index 5f0c0ede..0686c755 100644 --- a/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/main.go +++ b/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/main.go @@ -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) } diff --git a/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/main_test.go b/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/main_test.go index 31676cef..5c0c3c8a 100644 --- a/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/main_test.go +++ b/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/main_test.go @@ -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" { diff --git a/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/seatlist.go b/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/seatlist.go index 61e8d076..2c045320 100644 --- a/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/seatlist.go +++ b/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/seatlist.go @@ -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-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 + } +} diff --git a/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/seatlist_test.go b/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/seatlist_test.go index 8cf119c9..65e7e7e5 100644 --- a/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/seatlist_test.go +++ b/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/seatlist_test.go @@ -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") + } +} diff --git a/modules/mesh-delivery-builder/cmd/mesh-delivery-builder/main.go b/modules/mesh-delivery-builder/cmd/mesh-delivery-builder/main.go index 4aaa8bc2..28b1a2f2 100644 --- a/modules/mesh-delivery-builder/cmd/mesh-delivery-builder/main.go +++ b/modules/mesh-delivery-builder/cmd/mesh-delivery-builder/main.go @@ -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) diff --git a/modules/mesh-delivery-builder/cmd/mesh-delivery-builder/seatlist.go b/modules/mesh-delivery-builder/cmd/mesh-delivery-builder/seatlist.go index 7d96edc9..76a12f0a 100644 --- a/modules/mesh-delivery-builder/cmd/mesh-delivery-builder/seatlist.go +++ b/modules/mesh-delivery-builder/cmd/mesh-delivery-builder/seatlist.go @@ -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 + } +} diff --git a/modules/mesh-delivery-builder/cmd/mesh-delivery-builder/seatlist_test.go b/modules/mesh-delivery-builder/cmd/mesh-delivery-builder/seatlist_test.go new file mode 100644 index 00000000..a5e1d795 --- /dev/null +++ b/modules/mesh-delivery-builder/cmd/mesh-delivery-builder/seatlist_test.go @@ -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") + } +} diff --git a/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/input_integration_test.go b/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/input_integration_test.go index 91048da9..79cc3adc 100644 --- a/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/input_integration_test.go +++ b/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/input_integration_test.go @@ -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) diff --git a/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/main.go b/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/main.go index 0c37ff93..8acd43e2 100644 --- a/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/main.go +++ b/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/main.go @@ -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) diff --git a/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/seatlist.go b/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/seatlist.go index 8a2ebea9..759a039b 100644 --- a/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/seatlist.go +++ b/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/seatlist.go @@ -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 + } +} diff --git a/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/seatlist_test.go b/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/seatlist_test.go index 1773a17b..e12a1dba 100644 --- a/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/seatlist_test.go +++ b/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/seatlist_test.go @@ -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 } diff --git a/modules/mesh-delivery-pipeline/cmd/mesh-delivery-pipeline/main.go b/modules/mesh-delivery-pipeline/cmd/mesh-delivery-pipeline/main.go index 8e593380..024d800d 100644 --- a/modules/mesh-delivery-pipeline/cmd/mesh-delivery-pipeline/main.go +++ b/modules/mesh-delivery-pipeline/cmd/mesh-delivery-pipeline/main.go @@ -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 } diff --git a/modules/mesh-delivery-pipeline/cmd/mesh-delivery-pipeline/run_integration_test.go b/modules/mesh-delivery-pipeline/cmd/mesh-delivery-pipeline/run_integration_test.go index 293c9da3..1f3f9581 100644 --- a/modules/mesh-delivery-pipeline/cmd/mesh-delivery-pipeline/run_integration_test.go +++ b/modules/mesh-delivery-pipeline/cmd/mesh-delivery-pipeline/run_integration_test.go @@ -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) diff --git a/modules/mesh-delivery-pipeline/cmd/mesh-delivery-pipeline/verbs.go b/modules/mesh-delivery-pipeline/cmd/mesh-delivery-pipeline/verbs.go index a5816073..dbf008e0 100644 --- a/modules/mesh-delivery-pipeline/cmd/mesh-delivery-pipeline/verbs.go +++ b/modules/mesh-delivery-pipeline/cmd/mesh-delivery-pipeline/verbs.go @@ -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 diff --git a/modules/mesh-delivery-pipeline/cmd/mesh-delivery-pipeline/verbs_test.go b/modules/mesh-delivery-pipeline/cmd/mesh-delivery-pipeline/verbs_test.go index 9cb87105..6368b5cd 100644 --- a/modules/mesh-delivery-pipeline/cmd/mesh-delivery-pipeline/verbs_test.go +++ b/modules/mesh-delivery-pipeline/cmd/mesh-delivery-pipeline/verbs_test.go @@ -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") + } +} diff --git a/modules/mesh-delivery-pipeline/manifest_test.go b/modules/mesh-delivery-pipeline/manifest_test.go index 81f65253..ea6925a7 100644 --- a/modules/mesh-delivery-pipeline/manifest_test.go +++ b/modules/mesh-delivery-pipeline/manifest_test.go @@ -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. diff --git a/modules/mesh-delivery-pipeline/module.json b/modules/mesh-delivery-pipeline/module.json index 37a6d125..665b4fd9 100644 --- a/modules/mesh-delivery-pipeline/module.json +++ b/modules/mesh-delivery-pipeline/module.json @@ -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",