Let a process say what it does on the bus (issue 1179)
This commit is contained in:
jochen
@@ -1184,6 +1184,10 @@ func (r Resolution) compose(with Rendering, owner map[string]string,
|
||||
delete(copied, SecretsInEnvironment)
|
||||
delete(copied, NamesOnPurpose)
|
||||
delete(copied, TrustedField)
|
||||
// A process's bus stays controller-side (novox/hq ADR 0436 §1): never composed onto the
|
||||
// node-engine's process resource, whose wire declaration is mesh-host's own, vendored, with a
|
||||
// strict decoder that refuses an unknown key.
|
||||
delete(copied, ProcessBusField)
|
||||
// **An operator's value, from the assignment** (novox/hq ADR 0112, ADR 0155): what a
|
||||
// definition may not carry because it is true of one installation only. Filled from
|
||||
// the same layers a mergeable file takes, and refused when no layer set it.
|
||||
|
||||
@@ -1990,6 +1990,7 @@ func parseManifest(raw []byte, predefined PredefinedList) (Manifest, error) {
|
||||
// facts about the catalogue and are checked at registration (CatalogueProblems).
|
||||
problems = append(problems, declaredSeatProblems(m)...)
|
||||
problems = append(problems, RunsAsProblems(m)...)
|
||||
problems = append(problems, ProcessBusProblems(m)...)
|
||||
problems = append(problems, ReadsHolderProblems(m)...)
|
||||
problems = append(problems, FilesHolderProblems(m)...)
|
||||
if m.Computed != "" && len(m.Resources) > 0 {
|
||||
@@ -2855,12 +2856,7 @@ func replacesProblems(m Manifest) []string {
|
||||
func invokeProblems(m Manifest) []string {
|
||||
var problems []string
|
||||
for _, t := range m.Invokes {
|
||||
if t == "*" {
|
||||
continue
|
||||
}
|
||||
t = strings.TrimPrefix(t, "seat:")
|
||||
module, tool, named := strings.Cut(t, ".")
|
||||
if !named || !name.MatchString(module) || !toolName.MatchString(tool) {
|
||||
if !invokeNameValid(t) {
|
||||
problems = append(problems, fmt.Sprintf(
|
||||
"%s invokes %q, which does not name a tool: a module invokes <module>.<tool>, or "+
|
||||
"* for every tool on the mesh (novox/hq ADR 0152)", m.Module, t))
|
||||
|
||||
@@ -0,0 +1,189 @@
|
||||
package catalogue
|
||||
|
||||
import (
|
||||
"bytes"
|
||||
"encoding/json"
|
||||
"fmt"
|
||||
"strings"
|
||||
)
|
||||
|
||||
// ProcessBusField is the key a process resource declares its bus on. Controller-side only: never composed
|
||||
// onto the node-engine's process resource (novox/hq ADR 0436 §1) — the wire declaration is mesh-host's own,
|
||||
// vendored, with a strict decoder that refuses an unknown key.
|
||||
const ProcessBusField = "bus"
|
||||
|
||||
// ProcessBus is what one process of a module does on the bus (novox/hq ADR 0436 §1, §2, §3, §5): its own bus
|
||||
// account, granted exactly this and nothing of the module's other grants or the tool runner's — never a seat
|
||||
// to hold or serve, never a tool call or seat verb answered, never a restricted stream heard, never a
|
||||
// warrant published (§3, §5).
|
||||
type ProcessBus struct {
|
||||
// Emits are the events this process emits, named locally, as the module's own Emits names them
|
||||
// (EmitsAll): a process emits only what its module may emit as a whole.
|
||||
Emits []string `json:"emits,omitempty"`
|
||||
// Queues are the module's own work-queue streams (ADR 0436 §6) this process consumes.
|
||||
Queues []string `json:"queues,omitempty"`
|
||||
// State is the module's own state, by local name, this process reads and writes.
|
||||
State []string `json:"state,omitempty"`
|
||||
// Asks says whether this process reads the router's records of this module's own asks.
|
||||
Asks bool `json:"asks,omitempty"`
|
||||
// Invokes are the tools and seat verbs this process calls, each <module>.<tool> or a role's
|
||||
// seat:<seat>.<verb> (as Manifest.Invokes), or * for every tool on the mesh.
|
||||
Invokes []string `json:"invokes,omitempty"`
|
||||
}
|
||||
|
||||
// UnmarshalJSON reads a process's bus strictly: an unknown key — serves, a seat to hold or claim, or
|
||||
// anything else that would answer a tool call or a seat's verb — is refused here, by name (novox/hq ADR
|
||||
// 0436 §3).
|
||||
func (b *ProcessBus) UnmarshalJSON(raw []byte) error {
|
||||
type plain ProcessBus
|
||||
var p plain
|
||||
dec := json.NewDecoder(bytes.NewReader(raw))
|
||||
dec.DisallowUnknownFields()
|
||||
if err := dec.Decode(&p); err != nil {
|
||||
return fmt.Errorf("a process's bus is {emits, queues, state, asks, invokes}: %w", typedUnknown(err))
|
||||
}
|
||||
*b = ProcessBus(p)
|
||||
return nil
|
||||
}
|
||||
|
||||
// processBusOf is the bus a process resource declares, and whether it declared one at all.
|
||||
func processBusOf(r map[string]any) (bus ProcessBus, present bool, err error) {
|
||||
raw, present := r["bus"]
|
||||
if !present {
|
||||
return ProcessBus{}, false, nil
|
||||
}
|
||||
body, err := json.Marshal(raw)
|
||||
if err != nil {
|
||||
return ProcessBus{}, true, err
|
||||
}
|
||||
if err := json.Unmarshal(body, &bus); err != nil {
|
||||
return ProcessBus{}, true, err
|
||||
}
|
||||
return bus, true, nil
|
||||
}
|
||||
|
||||
// ProcessBuses is every process of this module that declares bus, by its id (the manifest's own resource
|
||||
// id, novox/hq ADR 0436, settled by the lead) — what the grant composition reads to grant a process
|
||||
// account exactly what it emits, consumes and calls (novox/hq ADR 0436 §2).
|
||||
func (m Manifest) ProcessBuses() map[string]ProcessBus {
|
||||
var out map[string]ProcessBus
|
||||
for _, r := range m.Resources {
|
||||
if fmt.Sprint(r["type"]) != "process" {
|
||||
continue
|
||||
}
|
||||
b, present, err := processBusOf(r)
|
||||
if !present || err != nil {
|
||||
continue
|
||||
}
|
||||
if out == nil {
|
||||
out = map[string]ProcessBus{}
|
||||
}
|
||||
out[fmt.Sprint(r["id"])] = b
|
||||
}
|
||||
return out
|
||||
}
|
||||
|
||||
// ProcessBusProblems is what is wrong with a manifest's processes' bus declarations (novox/hq ADR 0436 §1,
|
||||
// §3): the account rule is modelled on RunsAsProblems (seats_declared.go:522-550), and an invoked tool or
|
||||
// seat verb's shape on invokeProblems.
|
||||
func ProcessBusProblems(m Manifest) []string {
|
||||
var problems []string
|
||||
streams := map[string]bool{}
|
||||
for _, s := range m.Streams {
|
||||
streams[s.Name] = true
|
||||
}
|
||||
state := map[string]bool{}
|
||||
for _, s := range m.State {
|
||||
state[s.Name] = true
|
||||
}
|
||||
emits := map[string]bool{}
|
||||
for _, e := range m.EmitsAll() {
|
||||
emits[e] = true
|
||||
}
|
||||
for _, r := range m.Resources {
|
||||
if fmt.Sprint(r["type"]) != "process" {
|
||||
continue
|
||||
}
|
||||
id := fmt.Sprint(r["id"])
|
||||
if _, present := r["bus"]; !present {
|
||||
continue
|
||||
}
|
||||
b, _, err := processBusOf(r)
|
||||
if err != nil {
|
||||
problems = append(problems, fmt.Sprintf("%s's process %s declares bus: %v", m.Module, id, err))
|
||||
continue
|
||||
}
|
||||
user, _ := r["user"].(string)
|
||||
problems = append(problems, processAccountProblems(m, id, user)...)
|
||||
for _, e := range b.Emits {
|
||||
if !emits[e] {
|
||||
problems = append(problems, fmt.Sprintf(
|
||||
"%s's process %s's bus emits %q, which %s does not declare as its own (novox/hq ADR 0436 §1)",
|
||||
m.Module, id, e, m.Module))
|
||||
}
|
||||
}
|
||||
for _, q := range b.Queues {
|
||||
if !streams[q] {
|
||||
problems = append(problems, fmt.Sprintf(
|
||||
"%s's process %s's bus queues %q, which %s does not declare as its own stream (novox/hq "+
|
||||
"ADR 0436 §1)", m.Module, id, q, m.Module))
|
||||
}
|
||||
}
|
||||
for _, s := range b.State {
|
||||
if !state[s] {
|
||||
problems = append(problems, fmt.Sprintf(
|
||||
"%s's process %s's bus reads and writes state %q, which %s does not declare as its own "+
|
||||
"(novox/hq ADR 0436 §1)", m.Module, id, s, m.Module))
|
||||
}
|
||||
}
|
||||
for _, t := range b.Invokes {
|
||||
if !invokeNameValid(t) {
|
||||
problems = append(problems, fmt.Sprintf(
|
||||
"%s's process %s's bus invokes %q, which does not name a tool: a process invokes "+
|
||||
"<module>.<tool>, or a role's seat:<seat>.<verb> (novox/hq ADR 0436 §1)", m.Module, id, t))
|
||||
}
|
||||
}
|
||||
}
|
||||
return problems
|
||||
}
|
||||
|
||||
// processAccountProblems is what is wrong with one process's bus account (novox/hq ADR 0436 §1, §2): an
|
||||
// account the module makes — a `user` resource of that name — never root and never the operator account,
|
||||
// which a manifest cannot say safely (a node with no account has nothing to resolve it to, novox/hq ADR
|
||||
// 0436, settled by the lead).
|
||||
func processAccountProblems(m Manifest, processID, user string) []string {
|
||||
switch {
|
||||
case user == "":
|
||||
return []string{fmt.Sprintf(
|
||||
"%s's process %s declares bus and names no account; a process account is a user resource the "+
|
||||
"module makes, never root and never the operator account (novox/hq ADR 0436 §1, §2)",
|
||||
m.Module, processID)}
|
||||
case user == "root":
|
||||
return []string{fmt.Sprintf(
|
||||
"%s's process %s runs as root; a process account is the module's own, never root (novox/hq ADR "+
|
||||
"0436 §1, §2)", m.Module, processID)}
|
||||
case user == "${machine:account}":
|
||||
return []string{fmt.Sprintf(
|
||||
"%s's process %s runs as the operator account; a process account is the module's own, never the "+
|
||||
"operator's (novox/hq ADR 0436 §1, §2)", m.Module, processID)}
|
||||
}
|
||||
for _, r := range m.Resources {
|
||||
if fmt.Sprint(r["type"]) == "user" && fmt.Sprint(r["name"]) == user {
|
||||
return nil
|
||||
}
|
||||
}
|
||||
return []string{fmt.Sprintf(
|
||||
"%s's process %s runs as %s, which it does not make: a user resource named %s (novox/hq ADR 0436 §1, "+
|
||||
"§2)", m.Module, processID, user, user)}
|
||||
}
|
||||
|
||||
// invokeNameValid says a module or process's invoked entry names a tool: <module>.<tool>, a role's
|
||||
// seat:<seat>.<verb>, or * for every tool on the mesh (novox/hq ADR 0152, ADR 0436 §1).
|
||||
func invokeNameValid(t string) bool {
|
||||
if t == "*" {
|
||||
return true
|
||||
}
|
||||
t = strings.TrimPrefix(t, "seat:")
|
||||
first, second, named := strings.Cut(t, ".")
|
||||
return named && name.MatchString(first) && toolName.MatchString(second)
|
||||
}
|
||||
@@ -0,0 +1,188 @@
|
||||
package catalogue
|
||||
|
||||
import (
|
||||
"strings"
|
||||
"testing"
|
||||
)
|
||||
|
||||
// A process resource may declare `bus` (novox/hq ADR 0436 §1): what it emits, the module's own work-queue
|
||||
// streams it consumes, the module's own state it reads and writes, whether it reads the router's records of
|
||||
// this module's own asks, and the tools and seat verbs it calls. The fixture is mesh-mail's future manifest
|
||||
// shape (ADR 0436 §7, §8), copied from mesh-catalog/modules/mesh-mail/module.json and extended as settled by
|
||||
// the lead: process id "send" is the manifest's own `id` (internal/catalogue/runtime.go:72-74 is the model for
|
||||
// a process id); the account file uses the controller's real JSON tags (modules.go:1041-1056 issueWith).
|
||||
func mailSenderProcessManifest(t *testing.T, bus string) string {
|
||||
t.Helper()
|
||||
process := `{"id":"send","type":"process","run":["./mesh-mail","send"],"user":"mesh-mail-send"`
|
||||
if bus != "" {
|
||||
process += `,"bus":` + bus
|
||||
}
|
||||
process += `}`
|
||||
return `{"module":"mesh-mail","version":"1","emits":["mail.send.sent","mail.send.failed"],` +
|
||||
`"state":["sends"],"streams":[{"name":"mail-sends","events":["mail.send.approved"],"max-age-days":1,` +
|
||||
`"max-mib":64,"retention":"work-queue","max-deliveries":5,"duplicates":24}],` +
|
||||
`"resources":[{"id":"account","type":"user","name":"mesh-mail-send"},` + process + `]}`
|
||||
}
|
||||
|
||||
const mailSenderBus = `{"emits":["mail.send.sent","mail.send.failed"],"queues":["mail-sends"],` +
|
||||
`"state":["sends"],"asks":true,"invokes":["seat:operator-channel.tell"]}`
|
||||
|
||||
// 1179/AC1: a well-formed bus on a process whose account the module makes is accepted, and the catalogue
|
||||
// gives it whole to the grant composition (ProcessBuses, which 1180 reads).
|
||||
func TestAProcessDeclaresItsBusAndIsAccepted(t *testing.T) {
|
||||
m, err := ParseManifest([]byte(mailSenderProcessManifest(t, mailSenderBus)))
|
||||
if err != nil {
|
||||
t.Fatalf("a well-formed process bus was refused: %v", err)
|
||||
}
|
||||
buses := m.ProcessBuses()
|
||||
b, ok := buses["send"]
|
||||
if !ok {
|
||||
t.Fatalf("the process's bus was not given to the grant composition: %+v", buses)
|
||||
}
|
||||
if !strSliceEqual(b.Emits, []string{"mail.send.sent", "mail.send.failed"}) ||
|
||||
!strSliceEqual(b.Queues, []string{"mail-sends"}) || !strSliceEqual(b.State, []string{"sends"}) ||
|
||||
!b.Asks || !strSliceEqual(b.Invokes, []string{"seat:operator-channel.tell"}) {
|
||||
t.Fatalf("the bus read as %+v", b)
|
||||
}
|
||||
}
|
||||
|
||||
func strSliceEqual(a, b []string) bool {
|
||||
if len(a) != len(b) {
|
||||
return false
|
||||
}
|
||||
for i := range a {
|
||||
if a[i] != b[i] {
|
||||
return false
|
||||
}
|
||||
}
|
||||
return true
|
||||
}
|
||||
|
||||
// 1179/AC2: a bus on a process with no user, user root, or the operator account is refused naming the
|
||||
// process and the account rule.
|
||||
func TestAProcessBusAccountMustBeTheModulesOwn(t *testing.T) {
|
||||
for name, c := range map[string]struct{ user, says string }{
|
||||
"no user": {"", "send"},
|
||||
"user root": {"root", "root"},
|
||||
"the operator account": {"${machine:account}", "operator"},
|
||||
} {
|
||||
raw := `{"module":"mesh-mail","version":"1","emits":["mail.send.sent"],"state":["sends"],` +
|
||||
`"streams":[{"name":"mail-sends","events":["mail.send.approved"],"retention":"work-queue",` +
|
||||
`"max-deliveries":5}],"resources":[{"id":"account","type":"user","name":"mesh-mail-send"},` +
|
||||
`{"id":"send","type":"process","run":["./mesh-mail","send"]`
|
||||
if c.user != "" {
|
||||
raw += `,"user":"` + c.user + `"`
|
||||
}
|
||||
raw += `,"bus":{"emits":["mail.send.sent"]}}]}`
|
||||
_, err := ParseManifest([]byte(raw))
|
||||
if err == nil || !strings.Contains(err.Error(), "send") || !strings.Contains(err.Error(), c.says) {
|
||||
t.Errorf("%s: %v; want a refusal naming the process and %q", name, err, c.says)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// 1179/AC3: a bus naming a stream, state or event the module does not declare as its own is refused naming
|
||||
// it.
|
||||
func TestAProcessBusNamesOnlyTheModulesOwn(t *testing.T) {
|
||||
for name, c := range map[string]struct{ bus, says string }{
|
||||
"another module's event": {`{"emits":["billing.paid"]}`, "billing.paid"},
|
||||
"another module's queue": {`{"queues":["other-queue"]}`, "other-queue"},
|
||||
"another module's state": {`{"state":["other-state"]}`, "other-state"},
|
||||
} {
|
||||
_, err := ParseManifest([]byte(mailSenderProcessManifest(t, c.bus)))
|
||||
if err == nil || !strings.Contains(err.Error(), c.says) {
|
||||
t.Errorf("%s: %v; want a refusal naming %q", name, err, c.says)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// 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.
|
||||
func TestAProcessBusNeverServesOrHoldsASeat(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"},
|
||||
} {
|
||||
_, err := ParseManifest([]byte(mailSenderProcessManifest(t, c.bus)))
|
||||
if err == nil || !strings.Contains(err.Error(), c.says) {
|
||||
t.Errorf("%s: %v; want a refusal naming %q", name, err, c.says)
|
||||
}
|
||||
}
|
||||
accepted := `{"emits":["mail.send.sent","mail.send.failed"],"invokes":["seat:operator-channel.tell","mail.status"]}`
|
||||
if _, err := ParseManifest([]byte(mailSenderProcessManifest(t, accepted))); err != nil {
|
||||
t.Fatalf("a bus invoking a tool and a seat verb was refused: %v", err)
|
||||
}
|
||||
}
|
||||
|
||||
// 1179/AC5: a process declaring bus is composed with no bus key on the wire, and is otherwise byte for byte
|
||||
// what main composes for the same process without bus.
|
||||
func TestAProcessesBusNeverReachesTheWire(t *testing.T) {
|
||||
withBus, err := ParseManifest([]byte(mailSenderProcessManifest(t, mailSenderBus)))
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
withoutBus, err := ParseManifest([]byte(mailSenderProcessManifest(t, "")))
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
with := Rendering{Needed: map[string]map[string]string{}}
|
||||
outWith, err := Resolution{Node: "novox", Modules: []Manifest{withBus}}.Declaration(with)
|
||||
if err != nil {
|
||||
t.Fatalf("composing with bus: %v", err)
|
||||
}
|
||||
outWithout, err := Resolution{Node: "novox", Modules: []Manifest{withoutBus}}.Declaration(with)
|
||||
if err != nil {
|
||||
t.Fatalf("composing without bus: %v", err)
|
||||
}
|
||||
var withProcess, withoutProcess map[string]any
|
||||
for _, r := range outWith {
|
||||
if r["id"] == "mesh-mail.send" {
|
||||
withProcess = r
|
||||
}
|
||||
}
|
||||
for _, r := range outWithout {
|
||||
if r["id"] == "mesh-mail.send" {
|
||||
withoutProcess = r
|
||||
}
|
||||
}
|
||||
if withProcess == nil || withoutProcess == nil {
|
||||
t.Fatalf("the process was not composed at all: with=%v without=%v", withProcess, withoutProcess)
|
||||
}
|
||||
if _, has := withProcess["bus"]; has {
|
||||
t.Fatalf("the process resource carried a bus key onto the wire: %v", withProcess)
|
||||
}
|
||||
delete(withProcess, "bus")
|
||||
if !mapsEqual(withProcess, withoutProcess) {
|
||||
t.Fatalf("with bus declared, the process composed differently: %+v\nwant: %+v", withProcess, withoutProcess)
|
||||
}
|
||||
}
|
||||
|
||||
func mapsEqual(a, b map[string]any) bool {
|
||||
if len(a) != len(b) {
|
||||
return false
|
||||
}
|
||||
for k, v := range a {
|
||||
bv, ok := b[k]
|
||||
if !ok {
|
||||
return false
|
||||
}
|
||||
if av, ok := v.([]any); ok {
|
||||
bbv, ok := bv.([]any)
|
||||
if !ok || len(av) != len(bbv) {
|
||||
return false
|
||||
}
|
||||
for i := range av {
|
||||
if av[i] != bbv[i] {
|
||||
return false
|
||||
}
|
||||
}
|
||||
continue
|
||||
}
|
||||
if v != bv {
|
||||
return false
|
||||
}
|
||||
}
|
||||
return true
|
||||
}
|
||||
Reference in New Issue
Block a user