diff --git a/modules/mesh-delivery-pipeline/cmd/mesh-delivery-pipeline/approval.go b/modules/mesh-delivery-pipeline/cmd/mesh-delivery-pipeline/approval.go new file mode 100644 index 00000000..ff6cae89 --- /dev/null +++ b/modules/mesh-delivery-pipeline/cmd/mesh-delivery-pipeline/approval.go @@ -0,0 +1,107 @@ +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/process" + + "mesh-delivery-pipeline/internal/pipeline" +) + +// 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 (`.` 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. +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 +} + +// busAsker publishes the runner's approval ask on the seat operator-channel's accept subject under the runner's own name, its +// duplicate id the ask's own, so an ask published twice is kept once (§37). +type busAsker struct { + conn *nats.Conn + cred process.Account +} + +func (b busAsker) Ask(a oc.Ask) error { + data, err := json.Marshal(a) + if err != nil { + return err + } + msg := nats.NewMsg(oc.AcceptSubjectAsk(b.cred.Module)) + msg.Header.Set("x-event-id", a.ID) + msg.Header.Set("x-source", b.cred.Module) + msg.Header.Set("x-node", b.cred.Node) + msg.Header.Set("x-time", time.Now().UTC().Format(time.RFC3339Nano)) + msg.Header.Set("content-type", "application/json") + msg.Header.Set(nats.MsgIdHdr, a.ID) + msg.Data = data + return b.conn.PublishMsg(msg) +} + +var _ pipeline.Asker = busAsker{} + +// hearDecided wakes the runner on the operator-channel's decided for its own asks (§37, A-89): the warrant is read from the +// router's record, so an event the record does not back yet is heard again. +func hearDecided(hub *process.EventHub, h *pipeline.Hearer, module string, logf func(string, ...any)) error { + return hub.Handle(oc.SubjectDecided(module), func(m jetstream.Msg) { + var ev oc.Warrant + if err := json.Unmarshal(m.Data(), &ev); err != nil { + logf("%s holds what is no decided event, dropped: %v", m.Subject(), err) + _ = m.Term() + return + } + _, err := h.OnDecided(ev) + switch { + case errors.Is(err, pipeline.ErrRefused), errors.Is(err, pipeline.ErrNoAsk): + logf("the decided of ask %s takes nothing: %v", ev.Ask, err) + _ = m.Term() + case err != nil: + logf("the decided of ask %s is not taken yet, heard again: %v", ev.Ask, err) + _ = m.Nak() + default: + _ = m.Ack() + } + }) +} 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 ab75f8c3..a9167963 100644 --- a/modules/mesh-delivery-pipeline/cmd/mesh-delivery-pipeline/main.go +++ b/modules/mesh-delivery-pipeline/cmd/mesh-delivery-pipeline/main.go @@ -74,7 +74,13 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro return err } starter := &pipeline.Starter{Pipelines: createdKV{ctx: ctx, kv: pipelines}, Logf: logger.Printf, Say: say} - if err := hearStepEnded(hub, newHearer(ctx, js, cred, pipelines, kept, say, logger.Printf), logger.Printf); err != nil { + hearer := newHearer(ctx, js, cred, pipelines, kept, say, logger.Printf) + hearer.Asker = busAsker{conn: conn, cred: cred} + hearer.Where = routerRecordOf(cred.Module, calls.Call, calls.Ask) + if err := hearStepEnded(hub, hearer, logger.Printf); err != nil { + return err + } + if err := hearDecided(hub, hearer, cred.Module, logger.Printf); err != nil { return err } if err := hearPublished(ctx, hub, starter, calls, rv.Catalog, logger.Printf); err != nil { diff --git a/modules/mesh-delivery-pipeline/cmd/mesh-delivery-pipeline/router_record_test.go b/modules/mesh-delivery-pipeline/cmd/mesh-delivery-pipeline/router_record_test.go new file mode 100644 index 00000000..e56ce700 --- /dev/null +++ b/modules/mesh-delivery-pipeline/cmd/mesh-delivery-pipeline/router_record_test.go @@ -0,0 +1,37 @@ +package main + +import ( + "encoding/json" + "strings" + "testing" + + stdio "git.novox.be/novox/mesh-sdk/go" +) + +// Design 47 G §700, A-74: the router's record of an ask is read under the indexer's module's name from the state asks of the +// seat operator-channel, in the bucket of the seat's holder on the seat list, and carries the ask's warrant. +func TestTheRoutersRecordIsReadFromTheBucketOfTheSeatsHolder(t *testing.T) { + var asked []string + call := func(key string, body any) (json.RawMessage, error) { + asked = append(asked, key) + return json.RawMessage(`{"seats":[{"seat":"node-packet-filter","scope":"node","node":"n1","holders":[{"module":"pf","node":"n1"}]},` + + `{"seat":"operator-channel","scope":"mesh","holders":[{"module":"ops-router","node":"n2"}]}],"holds":2}`), nil + } + var bucket, id string + record := func(b, i string) (*stdio.StateEntry, error) { + bucket, id = b, i + return &stdio.StateEntry{Key: "mesh-delivery-pipeline." + i, Value: json.RawMessage(`{"state":"chosen","warrant":{"ask":"a1","asker":"mesh-delivery-pipeline","choice":"approve"}}`)}, nil + } + state, w, err := routerRecordOf("mesh-delivery-pipeline", call, record)("a1") + if err != nil || state != "chosen" || w == nil || w.Ask != "a1" { + t.Fatalf("Where = %q, %+v, %v; want the state chosen and the warrant of a1", state, w, err) + } + if bucket != "ops-router_asks" || id != "a1" || len(asked) != 1 || asked[0] != "seat:mesh-module-assigner.assignments" { + t.Errorf("read %q key %q after asking %v; want the holder's bucket ops-router_asks, the id, one question of the seat mesh-module-assigner", bucket, id, asked) + } + // No holder on the seat list: nothing is read, and the warrant stays unapplied with the reason said. + none := func(string, any) (json.RawMessage, error) { return json.RawMessage(`{"seats":[],"holds":0}`), nil } + if _, _, err := routerRecordOf("mesh-delivery-pipeline", none, record)("a1"); err == nil || !strings.Contains(err.Error(), "no module holds the seat operator-channel") { + t.Errorf("with no holder Where answered %v; want the reason", err) + } +} diff --git a/modules/mesh-delivery-pipeline/go.mod b/modules/mesh-delivery-pipeline/go.mod index 50178e66..ca25c2c2 100644 --- a/modules/mesh-delivery-pipeline/go.mod +++ b/modules/mesh-delivery-pipeline/go.mod @@ -9,6 +9,7 @@ require ( ) require ( + git.novox.be/novox/mesh-contracts/go/seats/meshmoduleassigner v0.0.0-00010101000000-000000000000 git.novox.be/novox/mesh-contracts/health-and-repair/go/clients/meshhandactsclient v0.0.0-00010101000000-000000000000 git.novox.be/novox/mesh-contracts/health-and-repair/go/seats/meshconditions v0.0.0-00010101000000-000000000000 git.novox.be/novox/mesh-delivery/contracts/go/clients/meshcatalogclient v0.0.0-00010101000000-000000000000 diff --git a/modules/mesh-delivery-pipeline/manifest_test.go b/modules/mesh-delivery-pipeline/manifest_test.go index c60c345f..f5afdd8f 100644 --- a/modules/mesh-delivery-pipeline/manifest_test.go +++ b/modules/mesh-delivery-pipeline/manifest_test.go @@ -68,7 +68,10 @@ func TestManifest(t *testing.T) { // 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", // RN3 (§53, §334): a publication's dependents behind. - "seat:mesh-catalog.outdated"})}, + "seat:mesh-catalog.outdated", + // A-74, A-89: who holds the seat operator-channel, whose state asks holds the router's record of the approval ask. + "seat:mesh-module-assigner.assignments"})}, + {"uses operator-channel", equalSets(m.Uses, []string{"operator-channel"})}, {"first-view tools", equalSets(m.Tools, []string{"pipeline_first_views", "pipeline_first_view_again"})}, // RN1/AC5 (§95, §130): the new assigner is delivered before the runner. {"program needs the assigner's program", slices.Equal(program.Needs, []string{"mesh-delivery-assigner:program"})}, diff --git a/modules/mesh-delivery-pipeline/module.json b/modules/mesh-delivery-pipeline/module.json index d0055720..f6ab3d52 100644 --- a/modules/mesh-delivery-pipeline/module.json +++ b/modules/mesh-delivery-pipeline/module.json @@ -43,7 +43,11 @@ "seat:mesh-catalog.registrations", "seat:mesh-catalog.revision", "seat:mesh-hand-acts.hand-act", - "seat:mesh-catalog.outdated" + "seat:mesh-catalog.outdated", + "seat:mesh-module-assigner.assignments" + ], + "uses": [ + "operator-channel" ], "state": [ {