diff --git a/cmd/mesh-controller/modules.go b/cmd/mesh-controller/modules.go index 754bc044..030f2af0 100644 --- a/cmd/mesh-controller/modules.go +++ b/cmd/mesh-controller/modules.go @@ -926,13 +926,45 @@ func brokerReachableAt(ctx context.Context, inv *inventory.Inventory, known brok // A module that runs its machine's provisioner client and declares no broker is issued the client's account // alone (issue 917): the client speaks on the bus only as itself. func mayIssue(m catalogue.Manifest) error { - if _, reads := m.OwnSecrets["broker"]; !reads && !runsTheProvisionerClient(m) { + processes := m.ProcessBuses() + // A module whose own tools run in the machine's tool runner — mesh-mail's real shape — declares no + // `broker` of its own at all (novox/hq ADR 0436 §2): its process's account is a different account, + // delivered as its own secret below, so a process with a bus is reason enough to pass this check on + // its own, before the module's own `broker` is asked for. + if _, reads := m.OwnSecrets["broker"]; !reads && !runsTheProvisionerClient(m) && len(processes) == 0 { return fmt.Errorf("%s declares no own secret named broker, so there is nowhere to deliver "+ "an account; a module that speaks on the bus declares \"own-secrets\": {\"broker\": }", m.Module) } + // And every process with a bus declaration (novox/hq ADR 0436 §2): its own account is delivered as the + // module's own secret broker-, never the module's `broker`, so a process with a bus and no + // such secret has nowhere to read an account either. + for id := range processes { + name := processSecretName(id) + if _, reads := m.OwnSecrets[name]; !reads { + return fmt.Errorf("%s's process %s declares bus, and declares no own secret named %s, so there "+ + "is nowhere to deliver its account; a process with a bus declares \"own-secrets\": "+ + "{%q: } (novox/hq ADR 0436 §2)", m.Module, id, name, name) + } + } return nil } +// processSecretName is the own secret a process's bus account is delivered in (novox/hq ADR 0436 §2): +// broker-, never the module's own `broker`. +func processSecretName(processID string) string { return "broker-" + processID } + +// processPrincipalFor is the one process of m's bus declarations named processID, as `rollout mint` finds it +// again from a KindProcess principal's bare id (novox/hq ADR 0436 §1, §2): false when the manifest no longer +// declares it. +func processPrincipalFor(m catalogue.Manifest, processID string) (broker.ProcessPrincipal, bool) { + for _, pr := range inventory.ProcessesOf(m) { + if pr.Process == processID { + return pr, true + } + } + return broker.ProcessPrincipal{}, false +} + // issueOnTheNewBus gives an assigned module its credential on the bus being built. // // **Three things differ from a management call, and each is the point of the move.** The credential @@ -968,12 +1000,82 @@ func issueOnTheNewBus(ctx context.Context, inv *inventory.Inventory, m catalogue return err } } + // Every process with a bus declaration gets an account of its own (novox/hq ADR 0436 §1, §2), minted and + // delivered whether or not the module itself speaks on the bus — its own secret is broker-, + // never the module's `broker`. + for _, pr := range inventory.ProcessesOf(m) { + if err := issueProcess(ctx, inv, m, pr, node, busAddress, known, reachable); err != nil { + return err + } + } if !runsTheProvisionerClient(m) { return nil } return issueProvisionerClient(ctx, inv, m, node, known, reachable) } +// issueProcess gives one process its account: minted under its own name .., +// delivered as the module's own secret broker- in the provisioner client's form (novox/hq ADR +// 0436 §2), handed to the vault as any covered account is, and its work-queue consumer or consumers +// created where the bus can be reached — never the module's claims, since a process holds no seat. +func issueProcess(ctx context.Context, inv *inventory.Inventory, m catalogue.Manifest, pr broker.ProcessPrincipal, + node, busAddress string, known broker.Broker, reachable string) error { + user := broker.Principal{Kind: broker.KindProcess, Node: node, Module: m.Module, Process: pr.Process}.Username() + password, err := inv.MintBusPassword(ctx, inventory.BusUser{ + Username: user, Kind: inventory.BusProcess, Node: node, Module: m.Module, + }) + if err != nil { + return err + } + name := processSecretName(pr.Process) + held, err := heldProcessAccount(known, reachable, node, m.Module, user, password) + if err != nil { + return err + } + if err := inv.AcceptSecretForModule(ctx, node, m.Module, name, string(held)); err != nil { + return err + } + if err := handIssuedToVault(ctx, inv, m, node, name, string(held)); err != nil { + return err + } + if len(pr.Queues) > 0 { + if busAddress == "" { + fmt.Printf(" %s's process %s consumes its work queue; its consumer is created when the bus is "+ + "reachable (`push --behind`, then `rollout mint` again is harmless)\n", m.Module, pr.Process) + } else { + js, err := broker.Dial(busAddress) + if err != nil { + return fmt.Errorf("the credential is minted and the mesh cannot reach the bus to create "+ + "%s's process %s's work-queue consumer: %w", m.Module, pr.Process, err) + } + defer js.Close() + who := broker.Principal{Kind: broker.KindProcess, Node: node, Module: m.Module, Process: pr.Process} + for _, q := range pr.Queues { + if err := js.EnsureConsumer(broker.QueueConsumer(who, q)); err != nil { + return err + } + } + } + } + fmt.Printf("bus user %s minted for %s's process %s, scoped to its bus declaration\n", user, m.Module, pr.Process) + fmt.Printf(" sealed to %s as %s. It arrives with the next push — `push %s` to send it\n", node, name, node) + return nil +} + +// heldProcessAccount is a process's account, in the provisioner client's form (novox/hq ADR 0436 §2): +// {url, fingerprint, node, module, user, password} — never claims, since a process holds no seat (§3). +func heldProcessAccount(known broker.Broker, reachable, node, module, user, password string) ([]byte, error) { + return json.Marshal(struct { + URL string `json:"url"` + Fingerprint string `json:"fingerprint,omitempty"` + Node string `json:"node"` + Module string `json:"module"` + User string `json:"user"` + Password string `json:"password"` + }{URL: "nats://" + reachable, Fingerprint: known.Fingerprint, Node: node, Module: module, User: user, + Password: password}) +} + // runsTheProvisionerClient says a module runs its machine's provisioner client, which reaches the bus on an // account of its own (novox/hq ADR 0349 clause 11, design 57 P3): it claims the seat and declares the own secret // that account is delivered in. The same test the bus's user list makes (inventory's declaredFor). diff --git a/cmd/mesh-controller/process_issue_integration_test.go b/cmd/mesh-controller/process_issue_integration_test.go new file mode 100644 index 00000000..be013ced --- /dev/null +++ b/cmd/mesh-controller/process_issue_integration_test.go @@ -0,0 +1,87 @@ +//go:build integration + +package main + +import ( + "context" + "strings" + "testing" + + "github.com/novox/mesh-controller/internal/broker" +) + +// review round 3, fault B (1181): the test asserted a literal url, while issueOnTheNewBus works out the +// machine's reachable address from the catalogue's network records (brokerReachableAt), not from this +// process's environment — the real try found "nats://anchor.internal:5671", not the literal guessed here. +// Fixed by calling issueProcess directly, with the known broker and reachable address given to it, exactly +// as issued_to_vault_test.go's issue/issueWith and issueProvisionerClient are called directly; the url +// asserted below is one of those two inputs, never a separate guess. +// +// review round 1, findings 3 and 4 (1181/AC1, AC2): module issue mints and delivers a process's own +// account against the store — no bus to dial, busAddress "" — for mesh-mail's real manifest shape (review +// round 1 finding 1: no own secret `broker` at all, its tools served by the tool runner), and issuing it +// again is the path every bus account takes (`rollout mint`'s KindProcess case, and a re-run of `module +// issue`): it rotates the password, and the old hash no longer matches. +func TestModuleIssueMintsAndDeliversAProcesssAccount(t *testing.T) { + open := aMesh(t) + ctx := t.Context() + inv := open.inventory + theVaultIsInstalled(t, open) + + m := mailSenderWithProcess() + m.Moved = true // so handIssuedToVault's accountIsCovered is true, and the hand-off is exercised whole + register(t, open, m) + + vault := newStandInVault(t) + vaultAskerForIssue = func(context.Context) vaultAsker { return vault.ask } + t.Cleanup(func() { vaultAskerForIssue = func(context.Context) vaultAsker { return newStandInVault(nil).ask } }) + + pr, found := processPrincipalFor(m, "send") + if !found { + t.Fatal("mesh-mail declares no process send") + } + known := broker.Broker{Fingerprint: "ab"} + const reachable = "10.0.0.1:4222" + issue := func() { + t.Helper() + if err := issueProcess(ctx, inv, m, pr, "laptop", "", known, reachable); err != nil { + t.Fatal(err) + } + } + + issue() + user := broker.Principal{Kind: broker.KindProcess, Node: "laptop", Module: "mesh-mail", Process: "send"}.Username() + if user != "laptop.mesh-mail.send" { + t.Fatalf("the process's username: %q", user) + } + hash1, held, err := inv.BusUserHash(ctx, user) + if err != nil || !held || hash1 == "" { + t.Fatalf("the process's account was not minted: held %v, %v", held, err) + } + + // 1181/AC1: accepted with the provisioner client's JSON, and handed to the vault whole — the url and + // the fingerprint being exactly what issueProcess was given above, not a guess at what it computes. + taken := vault.taken["accept"] + for _, want := range []string{`"url":"nats://` + reachable + `"`, `"fingerprint":"ab"`, `"node":"laptop"`, + `"module":"mesh-mail"`, `"user":"laptop.mesh-mail.send"`, `"password"`} { + if !strings.Contains(taken, want) { + t.Errorf("the vault was handed %q, without %s", taken, want) + } + } + if !strings.Contains(vault.said(), `"name":"broker-send"`) { + t.Errorf("the vault was not told the secret's name broker-send: %s", vault.said()) + } + // The file's own owner (mesh-mail-send, review round 1 finding 7's area) is 1170's mechanism, proven + // generically there and for this fixture in TestAProcessesOwnSecretIsOwnedByItsOwnAccount. + + // 1181/AC2: issued again — the path `rollout mint`'s KindProcess case and a re-run of `module issue` + // both take — rotates the password; the old hash no longer matches what is recorded. + issue() + hash2, held, err := inv.BusUserHash(ctx, user) + if err != nil || !held { + t.Fatalf("the process's account is gone after a re-issue: held %v, %v", held, err) + } + if hash1 == hash2 { + t.Fatal("issuing the process's account again did not rotate its password") + } +} diff --git a/cmd/mesh-controller/process_issue_test.go b/cmd/mesh-controller/process_issue_test.go new file mode 100644 index 00000000..bf4d02b3 --- /dev/null +++ b/cmd/mesh-controller/process_issue_test.go @@ -0,0 +1,106 @@ +package main + +import ( + "encoding/json" + "strings" + "testing" + + "github.com/novox/mesh-controller/internal/broker" + "github.com/novox/mesh-controller/internal/catalogue" +) + +// mailSenderWithProcess is mesh-mail's real manifest shape (novox/hq ADR 0436 §7, §8, the lead's fixture): +// no own secret named `broker` at all — its tools run in the machine's tool runner — and the process "send" +// declares bus, whose own secret broker-send is where module issue delivers its account (ADR 0436 §2), +// owned by mesh-mail-send, the user resource the process declares, at mode 0600 as every own secret is. +func mailSenderWithProcess() catalogue.Manifest { + m, err := catalogue.ParseManifest([]byte(`{"module":"mesh-mail","version":"1","emits":["mail.send.sent"], + "own-secrets":{"broker-send":{"path":"/var/lib/mesh-mail/broker-send","owner":"mesh-mail-send"}}, + "resources":[{"id":"account","type":"user","name":"mesh-mail-send"}, + {"id":"send","type":"process","run":["./mesh-mail","send"],"user":"mesh-mail-send", + "bus":{"emits":["mail.send.sent"]}}]}`)) + if err != nil { + panic(err) + } + return m +} + +// review round 1, finding 1 (1181): a module declaring no `broker` of its own at all — mesh-mail's real +// shape, tools served by the tool runner — still passes mayIssue on its process's bus declaration alone. +func TestAModuleWithNoBrokerAndOnlyAProcessMayBeIssued(t *testing.T) { + if err := mayIssue(mailSenderWithProcess()); err != nil { + t.Fatalf("a module with no broker of its own, whose process declares bus, was refused: %v", err) + } +} + +// 1181/AC1: a process with a bus declaration and no own secret broker- cannot be issued — the +// same refusal mayIssue already gives a module with no `broker` (novox/hq ADR 0436 §2). +func TestAProcessWithNoBrokerSecretCannotBeIssued(t *testing.T) { + m := mailSenderWithProcess() + delete(m.OwnSecrets, "broker-send") + err := mayIssue(m) + if err == nil { + t.Fatal("a process with no broker- own secret was issued an account") + } + for _, want := range []string{"mesh-mail", "send", "broker-send", "own-secrets"} { + if !strings.Contains(err.Error(), want) { + t.Errorf("the refusal does not say %q: %v", want, err) + } + } + if err := mayIssue(mailSenderWithProcess()); err != nil { + t.Errorf("a process declaring its own broker- secret was refused: %v", err) + } +} + +// 1181/AC1: the account held for a process is the provisioner client's form — {url, fingerprint, node, +// module, user, password} — named for the process, never the module's own account. +func TestAProcesssHeldAccountIsTheProvisionersForm(t *testing.T) { + user := broker.Principal{Kind: broker.KindProcess, Node: "novox", Module: "mesh-mail", Process: "send"}.Username() + if user != "novox.mesh-mail.send" { + t.Fatalf("the process's username: %q", user) + } + held, err := heldProcessAccount(broker.Broker{Fingerprint: "abc"}, "novox.example:4222", "novox", "mesh-mail", + user, "a-password") + if err != nil { + t.Fatal(err) + } + var got map[string]any + if err := json.Unmarshal(held, &got); err != nil { + t.Fatal(err) + } + want := map[string]any{"url": "nats://novox.example:4222", "fingerprint": "abc", "node": "novox", + "module": "mesh-mail", "user": "novox.mesh-mail.send", "password": "a-password"} + for k, v := range want { + if got[k] != v { + t.Errorf("%s: %v, want %v", k, got[k], v) + } + } + if len(got) != len(want) { + t.Errorf("the held account carries %v; want exactly %v (never claims — a process holds no seat)", got, want) + } +} + +// 1181/AC1: composed, the process's own secret belongs to its own account at 0600 (novox/hq ADR 0436 §1, §2; +// the owner mechanism is issue 1170's, proven generically there — this is mesh-mail's own shape). +func TestAProcessesOwnSecretIsOwnedByItsOwnAccount(t *testing.T) { + m := mailSenderWithProcess() + r := catalogue.Resolution{Node: "novox", Modules: []catalogue.Manifest{m}} + out, err := r.Declaration(catalogue.Rendering{ + Needed: map[string]map[string]string{"mesh-mail": {"broker-send": "sealed-account"}}, + }) + if err != nil { + t.Fatalf("composing the declaration: %v", err) + } + var found bool + for _, res := range out { + if res["type"] == "file" && res["path"] == "/var/lib/mesh-mail/broker-send" { + found = true + if got := res["owner"]; got != "mesh-mail-send" { + t.Fatalf("broker-send's owner: %v; want mesh-mail-send", got) + } + } + } + if !found { + t.Fatal("broker-send was not composed at all") + } +} diff --git a/cmd/mesh-controller/process_live_test.go b/cmd/mesh-controller/process_live_test.go new file mode 100644 index 00000000..35841eb4 --- /dev/null +++ b/cmd/mesh-controller/process_live_test.go @@ -0,0 +1,302 @@ +//go:build integration + +package main + +import ( + "errors" + "os" + "path/filepath" + "testing" + "time" + + "github.com/nats-io/nats-server/v2/server" + "github.com/nats-io/nats.go" + "github.com/nats-io/nats.go/jetstream" + + "github.com/novox/mesh-controller/internal/broker" + "github.com/novox/mesh-controller/internal/catalogue" + "github.com/novox/mesh-controller/internal/inventory" +) + +// mailSenderLiveManifest is mesh-mail's future manifest shape whole (novox/hq ADR 0436 §7, §8): the +// process "send" declares every clause of bus (emits, its one work-queue stream, its state, asks, and its +// invoked seat verb), so the account module issue delivers for it can be put through every grant at once. +func mailSenderLiveManifest(t *testing.T) catalogue.Manifest { + t.Helper() + m, err := catalogue.ParseManifest([]byte(`{"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":8, + "retention":"work-queue","max-deliveries":5}], + "own-secrets":{"broker-send":{"path":"/var/lib/mesh-mail/broker-send","owner":"mesh-mail-send"}}, + "resources":[{"id":"account","type":"user","name":"mesh-mail-send"}, + {"id":"send","type":"process","run":["./mesh-mail","send"],"user":"mesh-mail-send", + "bus":{"emits":["mail.send.sent","mail.send.failed"],"queues":["mail-sends"],"state":["sends"], + "asks":true,"invokes":["seat:operator-channel.tell"]}}]}`)) + if err != nil { + t.Fatal(err) + } + return m +} + +// operatorChannelWithTell is operator-channel as messenger declares it (broker.operatorChannel, novox/hq +// ADR 0259 §3), with the one verb this release adds: `tell` (novox/hq ADR 0436 §1's own fixture), so the +// process's invoked seat verb has a holder to answer it. +func operatorChannelWithTell() broker.Seat { + return broker.Seat{Name: "operator-channel", Scope: "mesh", Accepts: []string{"ask", "cancel"}, + Emits: []string{"decided"}, Serves: []string{"open", "history", "notify", "tell"}, + ByCaller: []string{"ask", "cancel", "decided"}, Records: []string{"messenger_asks"}} +} + +// review round 1, finding 5 (1181/AC3), the release's acceptance test, on a real bus and the store: the +// account module issue delivers for mesh-mail's process "send" — composed from the manifest the real way +// (inventory.ProcessesOf -> broker.Users -> broker.ComposeAccounts), minted with the password module issue +// mints (aMesh) — does exactly what its bus declares and nothing else. +func TestAProcessAccountOnARealBusAndStore(t *testing.T) { + open := aMesh(t) + ctx := t.Context() + inv := open.inventory + + m := mailSenderLiveManifest(t) + operatorChannel := operatorChannelWithTell() + mailSends := broker.ModuleStream{Module: "mesh-mail", Name: "MAIL_SENDS", Events: []string{"mail.send.approved"}, + MaxAgeDays: 1, MaxMiB: 8, WorkQueue: true, Duplicates: 1} + records := broker.Records{Nodes: []string{"laptop"}, Assigned: map[string][]broker.Declared{"laptop": { + {Module: broker.RuntimeModule}, + {Module: "messenger", Holds: []broker.Seat{operatorChannel}, RunsAs: "messenger"}, + {Module: "mesh-mail", NoAccount: true, Uses: []broker.Seat{operatorChannel}, Streams: []broker.ModuleStream{mailSends}, + Processes: inventory.ProcessesOf(m)}, + }}} + users, err := broker.Users(records) + if err != nil { + t.Fatal(err) + } + + // The four principals this test drives, each minted the password module issue mints — the same + // inv.MintBusPassword call issueProcess makes for the process account. mesh-mail publishes its own + // stream's events as the tool runner carrying it (review round 2's finding): the controller's own + // grants do not cover a module's stream, only the runtime that carries the module does. + var admin, runtime, messenger, sender broker.Principal + for _, p := range users { + switch { + case p.Kind == broker.KindController: + admin = p + case p.Kind == broker.KindNodeTools: + runtime = p + case p.Module == "messenger": + messenger = p + case p.Kind == broker.KindProcess: + sender = p + } + } + if admin.Kind == "" || runtime.Kind == "" || messenger.Kind == "" || sender.Kind == "" { + t.Fatalf("not every principal was composed: admin=%+v runtime=%+v messenger=%+v sender=%+v", + admin, runtime, messenger, sender) + } + if sender.Username() != "laptop.mesh-mail.send" { + t.Fatalf("the process's username: %q", sender.Username()) + } + + passwords := map[string]string{} + hashes := map[string]string{} + for _, p := range []struct { + principal broker.Principal + kind string + }{{admin, inventory.BusController}, {runtime, inventory.BusNodeTools}, {messenger, inventory.BusModule}, + {sender, inventory.BusProcess}} { + pw, err := inv.MintBusPassword(ctx, inventory.BusUser{Username: p.principal.Username(), Kind: p.kind, + Node: p.principal.Node, Module: p.principal.Module}) + if err != nil { + t.Fatal(err) + } + passwords[p.principal.Username()] = pw + hash, held, err := inv.BusUserHash(ctx, p.principal.Username()) + if err != nil || !held { + t.Fatalf("the minted password of %s was not recorded: %v, %v", p.principal.Username(), held, err) + } + hashes[p.principal.Username()] = hash + } + filled, missing := broker.WithPasswords([]broker.Principal{admin, runtime, messenger, sender}, hashes) + if len(missing) > 0 { + t.Fatalf("no password for %v", missing) + } + + accounts, err := broker.ComposeAccounts(filled) + if err != nil { + t.Fatal(err) + } + conf := filepath.Join(t.TempDir(), "bus.conf") + if err := os.WriteFile(conf, []byte("listen: 127.0.0.1:-1\njetstream { store_dir: "+ + `"`+t.TempDir()+`"`+" }\n"+accounts), 0o600); err != nil { + t.Fatal(err) + } + opts, err := server.ProcessConfigFile(conf) + if err != nil { + t.Fatalf("the composed accounts do not parse: %v", err) + } + opts.NoLog, opts.NoSigs = true, true + s, err := server.NewServer(opts) + if err != nil { + t.Fatal(err) + } + go s.Start() + if !s.ReadyForConnections(10 * time.Second) { + t.Fatal("the bus did not come up") + } + defer s.Shutdown() + + // Every connection names its own inbox prefix: the grant is `_INBOX..>`, and the client + // library's own default random inbox matches nothing granted, so replies are refused without it. + adminConn, err := broker.Dial(s.ClientURL(), nats.UserInfo(admin.Username(), passwords[admin.Username()]), + nats.CustomInboxPrefix("_INBOX."+admin.Username())) + if err != nil { + t.Fatal(err) + } + defer adminConn.Close() + if err := adminConn.EnsureStream(mailSends.Stream()); err != nil { + t.Fatal(err) + } + if err := adminConn.EnsureConsumer(broker.QueueConsumer(sender, sender.Queues[0])); err != nil { + t.Fatal(err) + } + if err := adminConn.EnsureBucket(broker.Bucket{Module: "mesh-mail", Name: "sends"}); err != nil { + t.Fatal(err) + } + js, err := jetstream.New(adminConn.Conn()) + if err != nil { + t.Fatal(err) + } + if _, err := js.CreateKeyValue(ctx, jetstream.KeyValueConfig{Bucket: "messenger_asks"}); err != nil { + t.Fatal(err) + } + + // The machine's tool runner, which carries mesh-mail (it has no account of its own) and so is the + // principal that publishes its own stream's events on its behalf. + runtimeConn, err := nats.Connect(s.ClientURL(), nats.UserInfo(runtime.Username(), passwords[runtime.Username()]), + nats.CustomInboxPrefix("_INBOX."+runtime.Username())) + if err != nil { + t.Fatal(err) + } + defer runtimeConn.Close() + + // messenger, the router, answers exactly its granted verbs — never a wildcard the grant does not give it. + holder, err := nats.Connect(s.ClientURL(), nats.UserInfo(messenger.Username(), passwords[messenger.Username()]), + nats.CustomInboxPrefix("_INBOX."+messenger.Username())) + if err != nil { + t.Fatal(err) + } + defer holder.Close() + heard := make(chan string, 1) + for _, verb := range []string{"open", "history", "notify", "tell"} { + if _, err := holder.Subscribe("mesh.seat.operator-channel.tool."+verb, func(m *nats.Msg) { + heard <- m.Subject + _ = m.Respond([]byte(`{"result":{}}`)) + }); err != nil { + t.Fatal(err) + } + } + _ = holder.Flush() + + refusals := make(chan error, 8) + sender2, err := nats.Connect(s.ClientURL(), nats.UserInfo(sender.Username(), passwords[sender.Username()]), + nats.CustomInboxPrefix("_INBOX."+sender.Username()), + nats.ErrorHandler(func(_ *nats.Conn, _ *nats.Subscription, err error) { refusals <- err })) + if err != nil { + t.Fatal(err) + } + defer sender2.Close() + + // 1. Its event. + if err := sender2.Publish("mesh.mod.mesh-mail.event.mail.send.sent", []byte("{}")); err != nil { + t.Fatal(err) + } + + // 2. Its state: writes and reads its own bucket. + senderKV, err := jetstream.New(sender2) + if err != nil { + t.Fatal(err) + } + bucket, err := senderKV.KeyValue(ctx, "mesh-mail_sends") + if err != nil { + t.Fatal(err) + } + if _, err := bucket.Put(ctx, "key", []byte("value")); err != nil { + t.Fatalf("the process may not write its own state: %v", err) + } + if got, err := bucket.Get(ctx, "key"); err != nil || string(got.Value()) != "value" { + t.Fatalf("the process may not read its own state back: %v, %v", got, err) + } + + // 3. Its queue: pulls and acknowledges MAIL_SENDS through its own consumer. Published as the tool + // runner carrying mesh-mail — the live path a module's own stream's event takes — never as the + // controller, whose own grants do not cover a module's stream at all. + runtimeJS, err := jetstream.New(runtimeConn) + if err != nil { + t.Fatal(err) + } + if _, err := runtimeJS.Publish(ctx, broker.ModuleStreamSubject("mesh-mail", "mail.send.approved"), []byte("{}")); err != nil { + t.Fatal(err) + } + reading, err := jetstream.New(sender2) + if err != nil { + t.Fatal(err) + } + consumer, err := reading.Consumer(ctx, "MAIL_SENDS", broker.QueueConsumer(sender, sender.Queues[0]).Name) + if err != nil { + t.Fatalf("the process may not bind its own queue consumer: %v", err) + } + batch, err := consumer.Fetch(1, jetstream.FetchMaxWait(3*time.Second)) + if err != nil { + t.Fatal(err) + } + got := false + for msg := range batch.Messages() { + got = true + if err := msg.Ack(); err != nil { + t.Fatalf("the process may not acknowledge its own queue: %v", err) + } + } + if !got { + t.Fatal("the process did not pull the message its own queue holds") + } + + // 4. Its invoked seat verb, in its own name. + if _, err := sender2.Request("mesh.seat.operator-channel.tool.tell", []byte(`{}`), 3*time.Second); err != nil { + t.Fatalf("the process may not call operator-channel.tell: %v", err) + } + select { + case s := <-heard: + if s != "mesh.seat.operator-channel.tool.tell" { + t.Fatalf("messenger heard %s", s) + } + case <-time.After(time.Second): + t.Fatal("messenger never heard the call") + } + + // 5. Refused, one at a time, so each refusal is counted against the attempt that caused it rather than + // any other: answering a tool call it holds no seat and serves no tool for. + if _, err := sender2.Subscribe("mesh.mod.mesh-mail.tool.>", func(*nats.Msg) {}); err != nil { + t.Fatal(err) + } + _ = sender2.Flush() + select { + case err := <-refusals: + if !errors.Is(err, nats.ErrPermissionViolation) { + t.Errorf("subscribing a tool call was not refused as a permissions violation: %v", err) + } + case <-time.After(3 * time.Second): + t.Error("subscribing a tool call the process does not serve was not refused") + } + // 6. Refused: publishing a warrant — only the holder, messenger, may say operator-channel decided. + if err := sender2.Publish("mesh.seat.operator-channel.event.decided.mesh-mail", []byte("{}")); err != nil { + t.Fatal(err) + } + _ = sender2.Flush() + select { + case err := <-refusals: + if !errors.Is(err, nats.ErrPermissionViolation) { + t.Errorf("publishing a warrant was not refused as a permissions violation: %v", err) + } + case <-time.After(3 * time.Second): + t.Error("publishing a warrant the process does not hold the seat for was not refused") + } +} diff --git a/cmd/mesh-controller/rollout.go b/cmd/mesh-controller/rollout.go index d1fb3b47..fbee3e1e 100644 --- a/cmd/mesh-controller/rollout.go +++ b/cmd/mesh-controller/rollout.go @@ -392,6 +392,32 @@ func rolloutMint(ctx context.Context, again bool) error { } modules++ + case broker.KindProcess: + // One process's own account (novox/hq ADR 0436 §1, §2), delivered as the module's own secret + // broker- — minted here exactly as `module issue` mints it, so a process issued + // once still gets a new password every time the bus's accounts are issued again. + m, inShelf := shelf[p.Module] + if !inShelf { + skipped++ + continue + } + pr, found := processPrincipalFor(m, p.Process) + if !found { + skipped++ + continue + } + name := processSecretName(p.Process) + if _, reads := m.OwnSecrets[name]; !reads { + fmt.Printf(" %s's process %s declares bus but declares no `%s` secret to receive a "+ + "credential in; skipped\n", p.Module, p.Process, name) + skipped++ + continue + } + if err := issueProcess(ctx, inv, m, pr, p.Node, "", known, busAddress); err != nil { + return err + } + modules++ + default: skipped++ } diff --git a/internal/broker/derived.go b/internal/broker/derived.go index c724b439..65c97048 100644 --- a/internal/broker/derived.go +++ b/internal/broker/derived.go @@ -190,18 +190,33 @@ func ConsumersOf(users []Principal) []ModuleConsumer { var out []ModuleConsumer seen := map[string]bool{} add := func(p Principal) { - c, needed := ConsumerFor(p) - if !needed || seen[p.Node+"/"+p.Module] { + key := p.Node + "/" + p.Module + if seen[key] { return } - seen[p.Node+"/"+p.Module] = true - out = append(out, ModuleConsumer{Node: p.Node, Module: p.Module, Consumer: c}) - // And one on each module stream it hears (novox/hq ADR 0363 §5). - for _, h := range p.Hears { - if len(h.Filters) == 0 { - continue + c, needed := ConsumerFor(p) + // **A module with a work-queue consumer of its own needs no other reason to be visited here** + // (novox/hq ADR 0436 §1, §2, §6): mesh-mail's sender consumes and watches nothing of its own, so + // ConsumerFor alone would have found nothing needed and skipped it before its one queue consumer + // was ever composed. + if !needed && len(p.Queues) == 0 { + return + } + seen[key] = true + if needed { + out = append(out, ModuleConsumer{Node: p.Node, Module: p.Module, Consumer: c}) + // And one on each module stream it hears (novox/hq ADR 0363 §5). + for _, h := range p.Hears { + if len(h.Filters) == 0 { + continue + } + out = append(out, ModuleConsumer{Node: p.Node, Module: p.Module, Consumer: heardConsumer(p, h)}) } - out = append(out, ModuleConsumer{Node: p.Node, Module: p.Module, Consumer: heardConsumer(p, h)}) + } + // And the one durable consumer of each of its own work-queue streams, for the one process its + // bus.queues names (novox/hq ADR 0436 §1, §2, §6). + for _, q := range p.Queues { + out = append(out, ModuleConsumer{Node: p.Node, Module: p.Module, Consumer: QueueConsumer(p, q)}) } } for _, p := range users { @@ -214,9 +229,13 @@ func ConsumersOf(users []Principal) []ModuleConsumer { continue } for _, d := range p.Carries { + // **A carried module's work-queue consumer is still the controller's to make** (novox/hq ADR + // 0436 §1, §2, §6), the same reason its ordinary consumer is (ADR 0198): mesh-mail has no + // account of its own (NoAccount) and is carried here, but its process does, and the consumer + // it reads through must exist and survive PruneHeard whether or not the module itself speaks. add(Principal{Kind: KindModule, Node: p.Node, Module: d.Module, Emits: d.Emits, Consumes: d.Consumes, Serves: d.Serves, Holds: d.Holds, Uses: d.Uses, Watches: d.Watches, - Streams: d.Streams, Hears: ComposedHears(d)}) + Streams: d.Streams, Hears: ComposedHears(d), Queues: d.Queues}) } } return out diff --git a/internal/broker/jetstream.go b/internal/broker/jetstream.go index 71fcb474..fa04d0b8 100644 --- a/internal/broker/jetstream.go +++ b/internal/broker/jetstream.go @@ -154,16 +154,10 @@ func Borrow(j *JetStream) *JetStream { return &lent } -// EnsureStream creates the stream if it is absent and brings it to match if it is present. -// -// **Idempotent, because the controller asserts on every start** rather than creating once at -// genesis: a stream somebody deleted, or a mesh raised from a restored backup, has to converge -// rather than run without the guarantee its messages assume. -// -// An update, not a delete and recreate. Recreating would discard every message the stream holds -// and every consumer's position in it — which for CONTROL means the pushes being held through a -// store restart, exactly the guarantee the stream exists for. -func (j *JetStream) EnsureStream(s Stream) error { +// StreamConfigFor is the server's own config for one of the mesh's streams, pure so the shape it builds — +// not least the restricted metadata a work-queue or a grant-restricted stream's dead letters age and hide +// by (novox/hq ADR 0436 §6, ADR 0376 §3) — is testable without a server. +func StreamConfigFor(s Stream) *nats.StreamConfig { want := &nats.StreamConfig{ Name: s.Name, Subjects: s.Subjects, @@ -174,8 +168,14 @@ func (j *JetStream) EnsureStream(s Stream) error { } want.AllowDirect = s.Direct want.AllowMsgTTL = s.AllowMsgTTL - if s.Restricted { - want.Metadata = map[string]string{RestrictedMetadata: "true"} + if s.Restricted || s.WorkQueue { + want.Metadata = map[string]string{} + if s.Restricted { + want.Metadata[RestrictedMetadata] = "true" + } + if s.WorkQueue { + want.Metadata[WorkQueueMetadata] = "true" + } } if s.MaxBytes > 0 { want.MaxBytes = s.MaxBytes @@ -193,6 +193,20 @@ func (j *JetStream) EnsureStream(s Stream) error { want.MaxMsgsPerSubject = 1 want.MaxAge = 0 } + return want +} + +// EnsureStream creates the stream if it is absent and brings it to match if it is present. +// +// **Idempotent, because the controller asserts on every start** rather than creating once at +// genesis: a stream somebody deleted, or a mesh raised from a restored backup, has to converge +// rather than run without the guarantee its messages assume. +// +// An update, not a delete and recreate. Recreating would discard every message the stream holds +// and every consumer's position in it — which for CONTROL means the pushes being held through a +// store restart, exactly the guarantee the stream exists for. +func (j *JetStream) EnsureStream(s Stream) error { + want := StreamConfigFor(s) switch _, err := j.js.StreamInfo(s.Name); { case err == nil: diff --git a/internal/broker/modstream_work_queue_test.go b/internal/broker/modstream_work_queue_test.go new file mode 100644 index 00000000..115bb973 --- /dev/null +++ b/internal/broker/modstream_work_queue_test.go @@ -0,0 +1,81 @@ +package broker + +import "testing" + +// mailSendsStream is mesh-mail's future `mail-sends` stream (novox/hq ADR 0436 §7, §8), as the bus holds it: +// work-queue retention, a 24-hour duplicate window, kept one day and at most 64 MiB. +var mailSendsStream = ModuleStream{Module: "mesh-mail", Name: "MAIL_SENDS", + Events: []string{"mail.send.approved"}, MaxAgeDays: 1, MaxMiB: 64, WorkQueue: true, Duplicates: 24} + +// 1177/AC1: a work-queue stream is composed with work-queue retention, refusing a new message when full +// (DiscardNew), a 24-hour duplicate window, a one-day maximum age and a 64 MiB cap. +func TestAWorkQueueStreamIsComposedToRefuseWhenFull(t *testing.T) { + s := mailSendsStream.Stream() + if s.Retention != RetentionWorkQueue { + t.Fatalf("retention: %v, want %v", s.Retention, RetentionWorkQueue) + } + if !s.DiscardNew { + t.Fatal("a work-queue stream did not refuse a new message when full") + } + if s.DuplicatesSeconds != 24*60*60 { + t.Fatalf("duplicate window: %d seconds, want %d (24 hours)", s.DuplicatesSeconds, 24*60*60) + } + if s.MaxAge != 24*60*60 { + t.Fatalf("max age: %d seconds, want %d (1 day)", s.MaxAge, 24*60*60) + } + if s.MaxBytes != 64<<20 { + t.Fatalf("max bytes: %d, want %d (64 MiB)", s.MaxBytes, 64<<20) + } +} + +// 1177/AC3: a stream with no work-queue clause composes exactly as main does: limits retention, dropping its +// oldest, never refusing a new message. +func TestANonWorkQueueStreamStillDropsItsOldest(t *testing.T) { + s := mailStream.Stream() + if s.Retention != RetentionLimits || s.DiscardNew || s.DuplicatesSeconds != 0 { + t.Fatalf("composed as %+v; want limits retention, never discarding new, no duplicate window", s) + } +} + +// 1178/AC1: the process a work-queue stream's bus.queues names gets exactly one durable consumer, +// __, with the stream's declared max-deliver — composed even though mesh-mail's +// sender otherwise consumes and watches nothing of its own, which alone would give ConsumerFor nothing +// needed. Explicit acknowledgement is EnsureConsumer's own rule for every consumer it makes (jetstream.go), +// so it is not re-asserted here. +func TestTheOneProcessAWorkQueuesBusNamesGetsItsOwnConsumer(t *testing.T) { + sender := Principal{Kind: KindModule, Node: "novox", Module: "mesh-mail", PasswordHash: "x", + Queues: []QueueOwner{{Process: "send", Stream: "MAIL_SENDS", MaxDeliver: 5}}} + got := ConsumersOf([]Principal{sender}) + if len(got) != 1 { + t.Fatalf("mesh-mail's consumers: %+v", got) + } + c := got[0].Consumer + if c.Name != "novox_mesh-mail_send" || c.Stream != "MAIL_SENDS" || c.MaxDeliver != 5 { + t.Fatalf("the queue's consumer: %+v", c) + } +} + +// review round 1, finding 2 (1178): mesh-mail's real shape has no account of its own (NoAccount) and is +// carried by the machine's tool runner — so its queue consumer has to come from the carried-module path in +// ConsumersOf, not only from a module with its own principal, or PruneHeard would remove what issueProcess +// made for it (novox/hq ADR 0436 §1, §2, §6). +func TestAWorkQueueConsumerSurvivesForAModuleTheRuntimeCarries(t *testing.T) { + users, err := Users(Records{Nodes: []string{"novox"}, Assigned: map[string][]Declared{"novox": { + {Module: RuntimeModule}, + {Module: "mesh-mail", NoAccount: true, + Queues: []QueueOwner{{Process: "send", Stream: "MAIL_SENDS", MaxDeliver: 5}}}, + }}}) + if err != nil { + t.Fatal(err) + } + got := ConsumersOf(users) + var found bool + for _, c := range got { + if c.Module == "mesh-mail" && c.Consumer.Name == "novox_mesh-mail_send" && c.Consumer.Stream == "MAIL_SENDS" { + found = true + } + } + if !found { + t.Fatalf("mesh-mail's queue consumer is missing from the carried module's composition: %+v", got) + } +} diff --git a/internal/broker/modstreams.go b/internal/broker/modstreams.go index fc584b67..3dc94a30 100644 --- a/internal/broker/modstreams.go +++ b/internal/broker/modstreams.go @@ -49,6 +49,12 @@ type ModuleStream struct { MaxMiB int // Restricted says a trusted config value of the module grants who hears it (ADR 0363 §4). Restricted bool + // WorkQueue says this stream holds ADR 0436 §6's work-queue retention: a message is removed once a + // consumer acknowledges it, and the bus refuses a new one when full rather than dropping its oldest. + WorkQueue bool + // Duplicates is the work-queue's duplicate window in hours (novox/hq ADR 0436 §6); meaningless unless + // WorkQueue. + Duplicates int } // Subjects are the subjects the stream takes. @@ -70,7 +76,11 @@ func (s ModuleStream) Stream() Stream { if s.Restricted { why += "; heard only by the consumers its trusted config value grants" } - return Stream{ + if s.WorkQueue { + why += "; a work-queue (novox/hq ADR 0436 §6): a message is removed once its one consumer " + + "acknowledges it, and a full stream refuses a new one rather than dropping what it holds" + } + out := Stream{ Name: s.Name, Subjects: s.Subjects(), Retention: RetentionLimits, @@ -80,6 +90,50 @@ func (s ModuleStream) Stream() Stream { // Said on the bus, so a dead letter kept from it ages out with it (novox/hq ADR 0376 §3). Restricted: s.Restricted, } + if s.WorkQueue { + out.Retention = RetentionWorkQueue + out.DiscardNew = true + out.DuplicatesSeconds = s.Duplicates * 60 * 60 + // Said on the bus too, reusing the one flag a restricted stream's dead letters already age and hide + // by (novox/hq ADR 0436 §6, ADR 0376 §3-4): a work-queue stream's grant is always empty (ADR 0436 + // §6 refuses one that carries a grant), so this never collides with a trusted-config restriction. + out.Restricted = true + // And its own metadata, apart from the bus's work-queue retention SEAT_* and other mesh-internal + // streams share without being one of this clause's: only this flag says a dead letter kept from the + // stream is drop-only (novox/hq ADR 0436 §6), never the stream's Retention alone. + out.WorkQueue = true + } + return out +} + +// QueueOwner is the one process of a module that is the durable consumer of one of its own work-queue +// streams (novox/hq ADR 0436 §1, §2, §6): named in that process's bus.queues, and checked at registration +// to be the only one (catalogue.WorkQueueStreamProblems). +type QueueOwner struct { + // Process is the process resource's id — its bus account's name, with the node and module (novox/hq ADR + // 0436 §2). + Process string + // Stream is the work-queue stream's name on the bus. + Stream string + // MaxDeliver bounds how many times a message is handed over before it is given up as a dead letter + // (novox/hq ADR 0436 §6). + MaxDeliver int +} + +// QueueConsumer is the durable consumer the one process a module's bus.queues names is granted on its own +// work-queue stream (novox/hq ADR 0436 §1, §2, §6): no filter, since it is the stream's only consumer — +// unlike heardConsumer, which is one of possibly several hearing a module's ordinary stream. +func QueueConsumer(p Principal, q QueueOwner) Consumer { + return Consumer{ + Name: p.Node + "_" + p.Module + "_" + q.Process, + Stream: q.Stream, + AckWaitSeconds: 30, + MaxDeliver: q.MaxDeliver, + Why: fmt.Sprintf("%s's process %s is the one consumer %s's bus.queues names for its work-queue "+ + "stream %s (novox/hq ADR 0436 §6); after max-deliver %d it gives a message up, and the "+ + "controller keeps it in %s, offered only drop", p.Module, q.Process, p.Module, q.Stream, + q.MaxDeliver, DeadLettersStream), + } } // ModuleStreams is every declared module stream as the controller asserts it, in name order. diff --git a/internal/broker/modstreams_workqueue_integration_test.go b/internal/broker/modstreams_workqueue_integration_test.go new file mode 100644 index 00000000..5e51b68f --- /dev/null +++ b/internal/broker/modstreams_workqueue_integration_test.go @@ -0,0 +1,101 @@ +//go:build integration + +package broker + +import ( + "testing" + "time" + + "github.com/nats-io/nats.go" + "github.com/nats-io/nats.go/jetstream" + + "github.com/novox/mesh-controller/internal/testbus" +) + +// 1178/AC4, on a real bus: the controller ensuring mesh-mail's work-queue stream and its one consumer gives +// the bus's own guarantees — a duplicate publish within the window is taken once, a message naked past +// max-deliver is not handed over again, and a full stream refuses a new publish (ADR 0436 §6). +// +// go test -tags integration ./internal/broker/ -run TestAWorkQueueStreamOnARealBus +func TestAWorkQueueStreamOnARealBus(t *testing.T) { + js, err := Dial(testbus.URL(t)) + if err != nil { + t.Fatal(err) + } + defer js.Close() + + // Small enough that a handful of messages fill it, so "the stream is full" is reachable within a test. + stream := ModuleStream{Module: "mesh-mail", Name: "MAIL_SENDS", Events: []string{"mail.send.approved"}, + MaxAgeDays: 1, MaxMiB: 1, WorkQueue: true, Duplicates: 1} + if err := js.EnsureStream(stream.Stream()); err != nil { + t.Fatal(err) + } + t.Cleanup(func() { _ = js.Context().DeleteStream(stream.Name) }) + + owner := QueueOwner{Process: "send", Stream: stream.Name, MaxDeliver: 5} + sender := Principal{Kind: KindModule, Node: "novox", Module: "mesh-mail", PasswordHash: "x"} + consumer := QueueConsumer(sender, owner) + consumer.AckWaitSeconds = 1 + if err := js.EnsureConsumer(consumer); err != nil { + t.Fatal(err) + } + + subject := ModuleStreamSubject("mesh-mail", "mail.send.approved") + + // A duplicate publish within the window is kept once (ADR 0436 §6's duplicate window). + publish := func(id string, size int) (*nats.PubAck, error) { + return js.Context().Publish(subject, make([]byte, size), nats.MsgId(id)) + } + if _, err := publish("dup-1", 16); err != nil { + t.Fatal(err) + } + if _, err := publish("dup-1", 16); err != nil { + t.Fatal(err) + } + if info, err := js.Context().StreamInfo(stream.Name); err != nil || info.State.Msgs != 1 { + t.Fatalf("a message published twice with the same id: %v msgs, err %v; want 1", info, err) + } + + // Naked five times, the consumer is not handed the message a sixth time. + api, err := jetstream.New(js.Conn()) + if err != nil { + t.Fatal(err) + } + reading, err := api.Consumer(t.Context(), stream.Name, consumer.Name) + if err != nil { + t.Fatal(err) + } + for i := 0; i < owner.MaxDeliver; i++ { + batch, err := reading.Fetch(1, jetstream.FetchMaxWait(2*time.Second)) + if err != nil { + t.Fatal(err) + } + got := false + for msg := range batch.Messages() { + got = true + _ = msg.Nak() + } + if !got { + t.Fatalf("delivery %d of %d never arrived", i+1, owner.MaxDeliver) + } + } + batch, err := reading.Fetch(1, jetstream.FetchMaxWait(2*time.Second)) + if err != nil { + t.Fatal(err) + } + for range batch.Messages() { + t.Fatalf("the message was delivered a %dth time, past its max-deliver of %d", owner.MaxDeliver+1, owner.MaxDeliver) + } + + // Full, the stream refuses a new publish rather than dropping what it holds (DiscardNew). + var fullErr error + for i := 0; i < 1<<16; i++ { + if _, err := publish("fill-"+time.Now().String(), 1<<16); err != nil { + fullErr = err + break + } + } + if fullErr == nil { + t.Fatal("a work-queue stream accepted every publish; want it to refuse one once full") + } +} diff --git a/internal/broker/nats.go b/internal/broker/nats.go index ecec2bba..71380f2d 100644 --- a/internal/broker/nats.go +++ b/internal/broker/nats.go @@ -48,6 +48,11 @@ const ( // service of the module holding NodeProvisionerSeat, outside the tool runner, under an account of its own so // the vault can tell it from an agent. Its authority is provisionerGrants and nothing else. KindProvisioner Kind = "provisioner" + // KindProcess is one process of a module that declares bus (novox/hq ADR 0436 §1, §2): a service of that + // module, outside the tool runner and outside the module's own account, under a bus account scoped to + // exactly what its bus says — never root, never the operator account, and never able to answer a tool + // call or a seat's verb (§3). One per process and machine. + KindProcess Kind = "process" ) // RuntimeModule is the module that IS the node's tool runtime (novox/hq ADR 0175). Where it is @@ -95,6 +100,9 @@ type Principal struct { Kind Kind Node string Module string + // Process is the process resource's id, for a KindProcess principal alone (novox/hq ADR 0436 §2): + // Username and inbox are ... + Process string Emits []string Consumes []string @@ -170,6 +178,12 @@ type Principal struct { // subjects it was granted (novox/hq ADR 0363). Streams []ModuleStream Hears []Heard + // Queues are its own work-queue streams and the one process each is the consumer of (novox/hq ADR 0436 + // §1, §2, §6); for a KindProcess principal, its own alone. + Queues []QueueOwner + // Asks says a KindProcess principal reads the records of its module's own asks (novox/hq ADR 0436 §1), + // on whichever seat of Uses keeps them (novox/hq ADR 0259 §3) — never set for any other kind. + Asks bool // SnapshotsTheBus is the bus's own module, the one holding mesh-broker (novox/hq ADR 0235). Its // whole authority is BusSnapshotGrants: it copies the streams for the night's backup and nothing @@ -446,6 +460,10 @@ func (p Principal) Username() string { // issues its credential through the same path a module's takes (`module issue`), and // that path knows the node and the module, not the kind. return p.Node + "." + p.Module + case KindProcess: + // One account per process and machine (novox/hq ADR 0436 §2): `..`, + // never the module's own name alone, which is a different account with a different secret. + return p.Node + "." + p.Module + "." + p.Process case KindNode: return "node." + p.Node case KindController: @@ -558,7 +576,7 @@ var BusSnapshotGrants = struct{ Publish []string }{Publish: []string{ // a permission that cannot be derived from a declaration is a permission nobody can explain. func PermissionsFor(p Principal) (Permissions, error) { for _, part := range []struct{ what, value string }{ - {"node", p.Node}, {"module", p.Module}, + {"node", p.Node}, {"module", p.Module}, {"process", p.Process}, } { if part.value == "" { continue @@ -735,6 +753,12 @@ func PermissionsFor(p Principal) (Permissions, error) { return Permissions{}, err } + case KindProcess: + var err error + if pub, sub, err = processGrants(p); err != nil { + return Permissions{}, err + } + case KindPerson: // Tools, and nothing else. Every subject a person may publish is a tool call; a person // who could publish an event would be able to claim a module said something. @@ -1128,6 +1152,13 @@ func PermissionsFor(p Principal) (Permissions, error) { // for the same reason a person has none: nothing is delivered to it. sub = append(sub, p.inbox()) } + if p.Kind == KindProcess { + // Its own reply space, and nothing wider (novox/hq ADR 0436 §1, §3): a process's queue consumer is + // pulled and acknowledged by publishing, granted above with its own stream and consumer name, never + // a generic EVENTS consumer's — consumerStream/consumerDurable below are a module's own, not a + // process's. + sub = append(sub, p.inbox()) + } if p.Kind == KindModule || p.Kind == KindNode || p.Kind == KindController { // Its own reply space, and nothing wider. diff --git a/internal/broker/process_grants.go b/internal/broker/process_grants.go new file mode 100644 index 00000000..9bbead83 --- /dev/null +++ b/internal/broker/process_grants.go @@ -0,0 +1,52 @@ +package broker + +// The bus account of one process of a module that declares bus (novox/hq ADR 0436 §1, §2, §3, §5). +// +// **Never the module's own account's grants, never the tool runner's.** A process's authority is read +// straight from its bus declaration — its emits, its queue's consumer, its state, whether it reads the +// records of its module's own asks, and the tools and seat verbs it calls — and nothing else: it holds no +// seat, serves no tool, hears no module stream, and answers nothing a caller asked (§3). Modelled on +// provisionerGrants, the other kind of process account the mesh mints (ADR 0349 clauses 6 and 11). + +// processGrants is one process's whole authority: to publish, and to subscribe. +func processGrants(p Principal) (pub, sub []string, err error) { + // 1. Its own events, under its module's own namespace — the same subjects the module's own principal + // would publish them on, had it declared them there instead (novox/hq ADR 0436 §1). + own := "mesh.mod." + p.Module + for _, e := range p.Emits { + pub = append(pub, own+".event."+e) + } + + // 2. The one durable consumer of each work-queue stream its bus names (novox/hq ADR 0436 §1, §2, §6): + // asked about, pulled from and acknowledged, by the same name ConsumersOf composes the consumer under + // (QueueConsumer) — and no other stream's. + for _, q := range p.Queues { + consumer := p.Node + "_" + p.Module + "_" + q.Process + pub = append(pub, + "$JS.API.CONSUMER.INFO."+q.Stream+"."+consumer, + "$JS.API.CONSUMER.MSG.NEXT."+q.Stream+"."+consumer, + "$JS.ACK."+q.Stream+"."+consumer+".>") + } + + // 3. Its module's own state, by the local names its bus names alone — never the whole of what the + // module keeps (novox/hq ADR 0436 §1), read and written as stateGrants gives any module's own bucket. + pub = append(pub, stateGrants(stateAccess{Module: p.Module, Node: p.Node, Keeps: p.State})...) + + // 4. The router's records of its module's own asks (novox/hq ADR 0436 §1, ADR 0259 §3): the same + // direct-get subjects a module's own principal reads for the seats its Uses names that keep them — + // read here from the one list, so the two cannot disagree. + if p.Asks { + pub = append(pub, SeatTrafficOf(p.Module, nil, p.Uses, nil).Records...) + } + + // 5. The tools and seat verbs it calls (novox/hq ADR 0152, ADR 0436 §1): the same derivation a + // module's own principal's invokes takes, so `callerNamed` (applied to every non-controller principal) + // adds the caller-named `call` variants for it exactly as it would for a module. + invoked, err := invokedSubjects(p.Invokes) + if err != nil { + return nil, nil, err + } + pub = append(pub, invoked...) + + return pub, sub, nil +} diff --git a/internal/broker/process_grants_test.go b/internal/broker/process_grants_test.go new file mode 100644 index 00000000..7866df30 --- /dev/null +++ b/internal/broker/process_grants_test.go @@ -0,0 +1,234 @@ +package broker + +import ( + "slices" + "sort" + "testing" +) + +// A module's process with a bus declaration gets a bus account of its own (novox/hq ADR 0436 §1, §2, §3, +// §5): named .., granted exactly its bus declaration and nothing of its +// module's other grants or the tool runner's. The fixture is mesh-mail's future manifest shape (ADR 0436 +// §7, §8): process "send", stream "mail-sends" (MAIL_SENDS), state "sends", using operator-channel as the +// router of its own asks (novox/hq ADR 0259 §3, seattraffic_test.go's operatorChannel()). + +// mailSenderProcess is the process principal PermissionsFor composes a bus account from, as +// inventory.processesOf would read it off mesh-mail's manifest (1179/AC1's fixture, 1178's queue owner). +func mailSenderProcess() Principal { + return Principal{Kind: KindProcess, Node: "novox", Module: "mesh-mail", Process: "send", + Emits: []string{"mail.send.sent", "mail.send.failed"}, + Queues: []QueueOwner{{Process: "send", Stream: "MAIL_SENDS", MaxDeliver: 5}}, + State: []string{"sends"}, Asks: true, Uses: []Seat{operatorChannel()}, + Invokes: []string{"seat:operator-channel.tell"}, + } +} + +// 1180/AC1: the process principal publishes its emits, pulls and acknowledges its one queue consumer, reads +// and writes its own state bucket, reads the router's records of its module's own asks, publishes its +// invoked seat verb in its own name and the plain form, and subscribes only its own inbox. +func TestAProcessIsGrantedExactlyItsBusDeclaration(t *testing.T) { + p := mailSenderProcess() + if got := p.Username(); got != "novox.mesh-mail.send" { + t.Fatalf("username: %q", got) + } + got := perms(t, p) + // review round 1, finding 8: the whole publish set, not merely that each wanted subject is somewhere in + // it — AC1 says "exactly". Dumped from a passing run (TestZZDumpProcessPublish, removed): 16 subjects, + // correct today. + wantPublish := []string{ + "$JS.ACK.MAIL_SENDS.novox_mesh-mail_send.>", + "$JS.API.CONSUMER.CREATE.KV_mesh-mail_sends.>", + "$JS.API.CONSUMER.DELETE.KV_mesh-mail_sends.>", + "$JS.API.CONSUMER.INFO.MAIL_SENDS.novox_mesh-mail_send", + "$JS.API.CONSUMER.MSG.NEXT.MAIL_SENDS.novox_mesh-mail_send", + "$JS.API.DIRECT.GET.KV_mesh-mail_sends.>", + "$JS.API.DIRECT.GET.KV_messenger_asks.$KV.messenger_asks.mesh-mail.>", + "$JS.API.STREAM.INFO.KV_mesh-mail_sends", + "$JS.FC.KV_mesh-mail_sends.>", + "$KV.mesh-mail_sends.>", + "mesh.mod.mesh-mail.event.mail.send.failed", + "mesh.mod.mesh-mail.event.mail.send.sent", + "mesh.seat.operator-channel.call.tell.*.novox~mesh-mail~send", + "mesh.seat.operator-channel.call.tell.novox~mesh-mail~send", + "mesh.seat.operator-channel.tool.tell", + "mesh.seat.operator-channel.tool.tell.*", + } + pub := append([]string(nil), got.Publish...) + sort.Strings(pub) + want := append([]string(nil), wantPublish...) + sort.Strings(want) + if !slices.Equal(pub, want) { + t.Fatalf("the process's whole publish set:\n%v\nwant exactly:\n%v", pub, want) + } + if !allowed(got.Subscribe, "_INBOX.novox.mesh-mail.send.x") { + t.Error("the process does not hear its own inbox") + } + if len(got.Subscribe) != 1 { + t.Errorf("the process subscribes %v; want only its own inbox", got.Subscribe) + } +} + +// 1180/AC3: a module with no bus process composes as main does, and adding one to mesh-mail changes +// neither the tool runner's grants for a module it carries nor another module's own. +func TestAddingAProcessLeavesOtherGrantsUnchanged(t *testing.T) { + plain := Principal{Kind: KindModule, Node: "novox", Module: "other", PasswordHash: "x"} + before := perms(t, plain) + users, err := Users(Records{Nodes: []string{"novox"}, Assigned: map[string][]Declared{"novox": { + {Module: "other"}, + {Module: "mesh-mail", Processes: []ProcessPrincipal{{Process: "send", + Emits: []string{"mail.send.sent"}}}, NoAccount: true}, + }}}) + if err != nil { + t.Fatal(err) + } + var sawProcess, sawOther bool + for _, u := range users { + if u.Kind == KindProcess && u.Module == "mesh-mail" { + sawProcess = true + } + if u.Module == "other" { + sawOther = true + if got := perms(t, u); !equalStringSlices(got.Publish, before.Publish) || !equalStringSlices(got.Subscribe, before.Subscribe) { + t.Errorf("another module's grants changed: %+v, want %+v", got, before) + } + } + if u.Kind == KindModule && u.Module == "mesh-mail" { + t.Error("mesh-mail has no own secret broker (NoAccount), and should have no module principal of its own") + } + } + if !sawProcess { + t.Fatal("the process account was not composed at all") + } + if !sawOther { + t.Fatal("the module with no process was not composed") + } +} + +// 1180/AC3: a process account is never carried by the machine's tool runner, so adding one to a module the +// runtime carries changes nothing of the runtime's own grants for that module. +func TestAddingAProcessLeavesTheToolRunnersCarryingUnchanged(t *testing.T) { + assigned := func(processes []ProcessPrincipal) map[string][]Declared { + return map[string][]Declared{"novox": { + {Module: RuntimeModule}, + {Module: "mesh-mail", Serves: []string{"status"}, Processes: processes}, + }} + } + without, err := Users(Records{Nodes: []string{"novox"}, Assigned: assigned(nil)}) + if err != nil { + t.Fatal(err) + } + with, err := Users(Records{Nodes: []string{"novox"}, Assigned: assigned([]ProcessPrincipal{ + {Process: "send", Emits: []string{"mail.send.sent"}}, + })}) + if err != nil { + t.Fatal(err) + } + runtimeOf := func(users []Principal) Principal { + for _, u := range users { + if u.Kind == KindNodeTools { + return u + } + } + t.Fatal("the tool runner was not composed") + return Principal{} + } + before, after := perms(t, runtimeOf(without)), perms(t, runtimeOf(with)) + if !equalStringSlices(before.Publish, after.Publish) || !equalStringSlices(before.Subscribe, after.Subscribe) { + t.Errorf("the tool runner's grants changed when mesh-mail gained a process: before %+v, after %+v", + before, after) + } +} + +// 1180/AC2, the grant test of ADR 0436 clause 3: a process account answers no tool call or seat verb, +// publishes no warrant, holds no seat (operator-channel included), and hears no restricted stream — however +// wide its own grants (emits, its queue, its state, its asks and its invokes) are. +func TestAProcessAccountAnswersNothing(t *testing.T) { + p := mailSenderProcess() + got := perms(t, p) + for _, s := range []string{ + // Every shape a tool call or a seat verb is answered on: a module's own namespace, a seat's, and + // the caller-named `call` kind naming any caller — which calledSubscribe would add if a `tool` + // grant ever reached Subscribe. + "mesh.mod.mesh-mail.tool.send", + "mesh.mod.mesh-mail.tool.>", + "mesh.seat.operator-channel.tool.tell", + "mesh.seat.operator-channel.call.tell.*", + "mesh.seat.operator-channel.call.tell.mesh-controller", + } { + if allowed(got.Subscribe, s) { + t.Errorf("a process account may subscribe %s, and could answer a call or a verb", s) + } + } + // No warrant: only operator-channel's holder publishes the asker's decided event (ADR 0259 §3). + if allowed(got.Publish, "mesh.seat.operator-channel.event.decided.mesh-mail") { + t.Error("a process account may publish a warrant") + } + // No seat held: the holder-only subjects (its accept queue, its worker) are refused. + for _, s := range []string{ + "mesh.seat.operator-channel.accept.ask.mesh-mail", + "$JS.API.CONSUMER.MSG.NEXT.SEAT_OPERATOR_CHANNEL.SEAT_OPERATOR_CHANNEL_worker", + } { + if allowed(got.Subscribe, s) || allowed(got.Publish, s) { + t.Errorf("a process account holds operator-channel: %s", s) + } + } + // No restricted stream: a process never hears a module stream at all (it has no Hears), restricted or + // not — the generic rule ADR 0376 §1 holds for it as it does for the tool runner. + if allowed(got.Subscribe, "mesh.stream.mesh-mail.event.mail.mesh.message.added") { + t.Error("a process account hears a module's stream") + } + if got.Subscribe == nil || len(got.Subscribe) != 1 { + t.Fatalf("a process account's whole subscribe list: %v; want only its own inbox", got.Subscribe) + } +} + +// Every process account composed for a mesh is held to the same rule (ADR 0436 clause 3): checked across +// the whole shape Users() composes, not only the one fixture above. +func TestEveryComposedProcessAccountAnswersNothing(t *testing.T) { + users, err := Users(Records{Nodes: []string{"novox"}, Assigned: map[string][]Declared{"novox": { + {Module: "messenger", Holds: []Seat{operatorChannel()}, RunsAs: "messenger"}, + {Module: "mesh-mail", RunsAs: "mesh-mail", Uses: []Seat{operatorChannel()}, + Processes: []ProcessPrincipal{{Process: "send", Emits: []string{"mail.send.sent"}, + Asks: true, Invokes: []string{"seat:operator-channel.tell"}}}}, + }}}) + if err != nil { + t.Fatal(err) + } + checked := 0 + for _, u := range users { + if u.Kind != KindProcess { + continue + } + checked++ + got := perms(t, u) + for _, s := range []string{ + "mesh.seat.operator-channel.call.tell.*", // answers any caller + "mesh.seat.operator-channel.accept.ask." + u.Module, // a caller's own submit, never a process's + "mesh.seat.operator-channel.event.decided." + u.Module, // the holder's warrant alone + } { + if allowed(got.Subscribe, s) || allowed(got.Publish, s) { + t.Errorf("%s may reach %s, which only a holder or an asking module's own principal may", + u.Username(), s) + } + } + } + if checked == 0 { + t.Fatal("no process account was composed to check") + } +} + +func equalStringSlices(a, b []string) bool { + if len(a) != len(b) { + return false + } + seen := map[string]bool{} + for _, x := range a { + seen[x] = true + } + for _, x := range b { + if !seen[x] { + return false + } + } + return true +} diff --git a/internal/broker/streams.go b/internal/broker/streams.go index 8be30cbb..d297d917 100644 --- a/internal/broker/streams.go +++ b/internal/broker/streams.go @@ -65,11 +65,21 @@ type Stream struct { // Restricted marks a module's stream restricted by a grant (novox/hq ADR 0363 §4), in the stream's own // metadata, so whoever keeps one of its messages reads it from the bus (ADR 0376 §3). Restricted bool + // WorkQueue marks a module stream declared retention: work-queue (novox/hq ADR 0436 §6), in the + // stream's own metadata, so a dead letter kept from it is read as drop-only from the bus, by the + // declaration and not by the bus's work-queue retention alone — which SEAT_* and other mesh-internal + // streams share without being one of ADR 0436's, and whose dead letters are still delivered again + // (ADR 0264 §4). + WorkQueue bool } // RestrictedMetadata is the key a restricted stream's metadata says it under. const RestrictedMetadata = "mesh-restricted" +// WorkQueueMetadata is the key a module stream declared retention: work-queue (novox/hq ADR 0436 §6) says +// it under: never set by a mesh-internal work-queue stream (a seat's, a check's), only by Stream.WorkQueue. +const WorkQueueMetadata = "mesh-work-queue" + // AssignmentsStream holds every assignment's membership, the newest per subject. const AssignmentsStream = "ASSIGNMENTS" diff --git a/internal/broker/users.go b/internal/broker/users.go index 9ad43933..46c01604 100644 --- a/internal/broker/users.go +++ b/internal/broker/users.go @@ -46,6 +46,9 @@ type Declared struct { Logs []Log // Streams are the streams it owns, each its events' one place (novox/hq ADR 0363). Streams []ModuleStream + // Queues are its own work-queue streams and the one process each is the consumer of (novox/hq ADR 0436 + // §1, §2, §6): one durable consumer per entry, named `__`. + Queues []QueueOwner // Hears are the module streams it hears on this machine, each with the subjects it was granted // (novox/hq ADR 0363 §5); HearsAStream says its `consumes` names a module that owns a stream, granted // or not, so its membership is issued again when a grant changes. @@ -73,6 +76,31 @@ type Declared struct { // declares the own secret ProvisionerSecret its account is delivered in (novox/hq ADR 0349 clause 11, design 57 // P3). The client is given a bus account of its own, never the module's or the tool runner's. ProvisionsNode bool + // Processes are this module's own processes that declare bus (novox/hq ADR 0436 §1), each getting a + // principal of its own: granted exactly what its bus says, and nothing of this module's other grants — + // composed whether or not the module itself has an account (NoAccount), because a process account is + // delivered in its own secret, broker-, never the module's `broker`. + Processes []ProcessPrincipal +} + +// ProcessPrincipal is one process of a module that declares bus (novox/hq ADR 0436 §1, §2), as the broker +// composes its own account from: its own emits, its own work-queue consumers, its own state, whether it +// reads the router's records of its module's own asks, and the tools and seat verbs it calls. +type ProcessPrincipal struct { + // Process is the process resource's id, naming the account ... + Process string + // Emits are the events this process may publish, named as the module's own Emits are. + Emits []string + // Queues are the module's own work-queue streams this process is the one consumer of. + Queues []QueueOwner + // State is the module's own state, by local name, this process reads and writes. + State []string + // Asks says whether this process reads the records of its module's own asks (novox/hq ADR 0259 §3), on + // whichever seats the module uses that keep them (the router's, operator-channel, among others). + Asks bool + // Invokes are the tools and seat verbs this process calls, each . or a role's + // seat:., or * for every tool on the mesh. + Invokes []string } // Records is what composing a user list needs to know about the mesh, and nothing more. @@ -137,6 +165,17 @@ func Users(r Records) ([]Principal, error) { } } for _, d := range r.Assigned[node] { + // **A process account is its own, delivered in its own secret** (broker-, novox/hq + // ADR 0436 §1, §2), never the module's `broker` — so it is composed whether or not the module + // itself has an account, and is never carried by the machine's runtime, which runs as the + // operator's account (ADR 0436 §5): added here, before either of those is judged. + for _, pr := range d.Processes { + out = append(out, Principal{ + Kind: KindProcess, Node: node, Module: d.Module, Process: pr.Process, + Emits: pr.Emits, Queues: pr.Queues, State: pr.State, Uses: d.Uses, Asks: pr.Asks, + Invokes: pr.Invokes, + }) + } if runtimeHere && d.Module == RuntimeModule { continue } @@ -155,7 +194,7 @@ func Users(r Records) ([]Principal, error) { Holds: d.Holds, Uses: d.Uses, Watches: d.Watches, Invokes: d.Invokes, State: stateNames(d.State), Reads: d.Reads, SnapshotsTheBus: d.SnapshotsTheBus, PerMachine: perMachineNames(d.State), KeyedReads: d.KeyedReads, - Logs: logNames(d.Logs), Streams: d.Streams, Hears: ComposedHears(d), + Logs: logNames(d.Logs), Streams: d.Streams, Hears: ComposedHears(d), Queues: d.Queues, }) // **The vault's holder asks the desks and the issuing providers itself** (issue 723), under the // account its own runtime runs on (ADR 0259 §8's runs-as): the one module granted it. diff --git a/internal/catalogue/declaration.go b/internal/catalogue/declaration.go index 6926dbea..a2d75976 100644 --- a/internal/catalogue/declaration.go +++ b/internal/catalogue/declaration.go @@ -680,7 +680,13 @@ func (r Resolution) compose(with Rendering, owner map[string]string, if !ok || family == "" || with.Needed[m.Module][name] == "" { continue } - first = append(first, ownedBy(m.SecretsOwner, map[string]any{ + // A per-secret owner (novox/hq ADR 0436 §1) stands in for the module's secrets-owner, when + // the manifest named one for this secret (or the family it belongs to). + owner := m.SecretsOwner + if own.Owner != "" { + owner = own.Owner + } + first = append(first, ownedBy(owner, map[string]any{ "id": NeedID(name), "type": "file", "path": own.Path, "sealed": with.Needed[m.Module][name], })) } @@ -694,6 +700,9 @@ func (r Resolution) compose(with Rendering, owner map[string]string, if m.Module == RuntimeModule && r.Account != "" { owner = r.Account } + if o := m.OwnSecrets[name].Owner; o != "" { + owner = o + } first = append(first, ownedBy(owner, map[string]any{ "id": NeedID(name), "type": "file", "path": m.OwnSecrets[name].Path, HandedSecretField: name, })) @@ -709,6 +718,9 @@ func (r Resolution) compose(with Rendering, owner map[string]string, if m.Module == RuntimeModule && r.Account != "" { owner = r.Account } + if o := m.OwnSecrets[name].Owner; o != "" { + owner = o + } first = append(first, ownedBy(owner, map[string]any{ "id": NeedID(name), "type": "file", "path": m.OwnSecrets[name].Path, VaultSealedField: box, })) @@ -742,6 +754,9 @@ func (r Resolution) compose(with Rendering, owner map[string]string, if m.Module == RuntimeModule && r.Account != "" { owner = r.Account } + if o := m.OwnSecrets[name].Owner; o != "" { + owner = o + } first = append(first, ownedBy(owner, map[string]any{ "id": NeedID(name), "type": "file", "path": m.OwnSecrets[name].Path, "sealed": sealed, })) @@ -1169,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. diff --git a/internal/catalogue/dir_into.go b/internal/catalogue/dir_into.go index 585358e3..dd3be2f5 100644 --- a/internal/catalogue/dir_into.go +++ b/internal/catalogue/dir_into.go @@ -168,7 +168,7 @@ func placedManifest(m Manifest, with Rendering) (Manifest, error) { if err != nil { return m, err } - own[name] = OwnSecret{Path: filled, Taken: s.Taken, IssuedBy: s.IssuedBy} + own[name] = OwnSecret{Path: filled, Taken: s.Taken, IssuedBy: s.IssuedBy, Prompt: s.Prompt, Owner: s.Owner} } m.OwnSecrets = own } diff --git a/internal/catalogue/manifest.go b/internal/catalogue/manifest.go index 3acc5cbd..85bbb067 100644 --- a/internal/catalogue/manifest.go +++ b/internal/catalogue/manifest.go @@ -1953,6 +1953,7 @@ func parseManifest(raw []byte, predefined PredefinedList) (Manifest, error) { problems = append(problems, LogProblems(m)...) // And the streams it owns, and who may hear them (streams.go, novox/hq ADR 0363). problems = append(problems, StreamProblems(m)...) + problems = append(problems, WorkQueueStreamProblems(m)...) // And the defaults it gives its settings (setting_defaults.go, novox/hq ADR 0262), or, written in the // config form, the whole declaration of each value (config.go, novox/hq ADR 0348). if m.DeclaresConfig() { @@ -1990,6 +1991,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 { @@ -2348,6 +2350,7 @@ func parseManifest(raw []byte, predefined PredefinedList) (Manifest, error) { m.Module, name, own.IssuedBy, IssuedOutside)) } problems = append(problems, ownSecretPromptProblems(m, name, own.Prompt, IsFamily(name))...) + problems = append(problems, ownSecretOwnerProblems(m, name, own.Owner)...) } localOf := map[string]string{} for _, to := range m.SecretRequirements() { @@ -2854,12 +2857,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 ., or "+ "* for every tool on the mesh (novox/hq ADR 0152)", m.Module, t)) @@ -2930,6 +2928,11 @@ type OwnSecret struct { // family's member suffix, filled where the secret is asked for. Optional; reviewed in the // module's pull request like the rest of the manifest. Prompt string + // Owner is the account this secret's files belong to on the machine, in place of the module's + // secrets-owner (novox/hq ADR 0436 §1): an account the module makes (a `user` resource of that + // name), never root and never the operator account. Empty means the module's secrets-owner, or + // root where that is absent too. + Owner string } // MeshMayMake says the mesh may put a value it makes in place of the one the secret holds: the @@ -2984,6 +2987,34 @@ func ownSecretPromptProblems(m Manifest, name, prompt string, family bool) []str return problems } +// ownSecretOwnerProblems is what is wrong with an own secret's declared owner (novox/hq ADR 0436 §1): naming +// an account the module makes — a `user` resource of that name — never root, which a secret file owned by it +// would make readable by anything, and never the operator account's placeholder (`${machine:account}`), which +// the mesh fills per machine and a manifest cannot say safely (novox/hq ADR 0436, settled by the lead). +func ownSecretOwnerProblems(m Manifest, name, owner string) []string { + if owner == "" { + return nil + } + if owner == "root" { + return []string{fmt.Sprintf( + "%s names %q as the owner of its secret %q; an own secret's owner is an account the module "+ + "makes, never root (novox/hq ADR 0436 §1)", m.Module, owner, name)} + } + if owner == "${machine:account}" { + return []string{fmt.Sprintf( + "%s names %q as the owner of its secret %q; an own secret's owner is an account the module "+ + "makes, never the operator account (novox/hq ADR 0436 §1)", m.Module, owner, name)} + } + for _, r := range m.Resources { + if fmt.Sprint(r["type"]) == "user" && fmt.Sprint(r["name"]) == owner { + return nil + } + } + return []string{fmt.Sprintf( + "%s names %q as the owner of its secret %q, which it does not make: a user resource named %q "+ + "(novox/hq ADR 0436 §1)", m.Module, owner, name, owner)} +} + // OwnSecrets is a module's own secrets by name. On the wire each is a path, or an object naming // the path and how it is taken; written back the way it was read, so a manifest the mesh holds // keeps its bytes. @@ -3006,13 +3037,14 @@ func (o *OwnSecrets) UnmarshalJSON(raw []byte) error { Taken string `json:"taken,omitempty"` IssuedBy string `json:"issued-by,omitempty"` Prompt string `json:"prompt,omitempty"` + Owner string `json:"owner,omitempty"` } dec := json.NewDecoder(bytes.NewReader(body)) dec.DisallowUnknownFields() if err := dec.Decode(&long); err != nil { - return fmt.Errorf("own-secrets.%s: a path, or {\"path\", \"taken\", \"issued-by\", \"prompt\"}: %w", name, typedUnknown(err)) + return fmt.Errorf("own-secrets.%s: a path, or {\"path\", \"taken\", \"issued-by\", \"prompt\", \"owner\"}: %w", name, typedUnknown(err)) } - out[name] = OwnSecret{Path: long.Path, Taken: long.Taken, IssuedBy: long.IssuedBy, Prompt: long.Prompt} + out[name] = OwnSecret{Path: long.Path, Taken: long.Taken, IssuedBy: long.IssuedBy, Prompt: long.Prompt, Owner: long.Owner} } *o = out return nil @@ -3021,7 +3053,7 @@ func (o *OwnSecrets) UnmarshalJSON(raw []byte) error { func (o OwnSecrets) MarshalJSON() ([]byte, error) { entries := make(map[string]any, len(o)) for name, s := range o { - if s.Taken == "" && s.IssuedBy == "" && s.Prompt == "" { + if s.Taken == "" && s.IssuedBy == "" && s.Prompt == "" && s.Owner == "" { entries[name] = s.Path continue } @@ -3035,6 +3067,9 @@ func (o OwnSecrets) MarshalJSON() ([]byte, error) { if s.Prompt != "" { long["prompt"] = s.Prompt } + if s.Owner != "" { + long["owner"] = s.Owner + } entries[name] = long } return json.Marshal(entries) @@ -3074,7 +3109,7 @@ func (o OwnSecrets) Lookup(name string) (s OwnSecret, family string, ok bool) { } if family == "" || len(declared) > len(family) { s, family, ok = OwnSecret{Path: strings.Replace(f.Path, "*", rest, 1), Taken: f.Taken, IssuedBy: f.IssuedBy, - Prompt: f.Prompt}, declared, true + Prompt: f.Prompt, Owner: f.Owner}, declared, true } } return s, family, ok diff --git a/internal/catalogue/own_secret_owner_test.go b/internal/catalogue/own_secret_owner_test.go new file mode 100644 index 00000000..fba354a0 --- /dev/null +++ b/internal/catalogue/own_secret_owner_test.go @@ -0,0 +1,116 @@ +package catalogue + +import ( + "strings" + "testing" +) + +// An own secret (or a family) may declare `owner` (novox/hq ADR 0436 §1): the account the files holding it +// belong to on the machine, when that account is not the module's secrets-owner default. The fixture is +// mesh-mail's future manifest shape (ADR 0436 §7): a process account `mesh-mail-send` the module makes, and its +// own-secret family `mail-credential-*` owned by it. + +// mailSenderManifest is the fixture every mail-sender issue builds on: the real parts of mesh-mail's manifest +// taken from mesh-catalog/modules/mesh-mail/module.json, extended as ADR 0436 §7-§8 describe, with `own` naming +// the account that owns the family's files. +func mailSenderManifest(t *testing.T, own string) Manifest { + t.Helper() + ownSecrets := `"mail-credential-*":{"path":"/var/lib/mesh-mail/mail-credential-*.secret","issued-by":"outside"` + if own != "" { + ownSecrets += `,"owner":"` + own + `"` + } + ownSecrets += `}` + m, err := ParseManifest([]byte(`{"module":"mesh-mail","version":"1","secrets-owner":"mesh-mail","own-secrets":{` + + ownSecrets + `},"resources":[{"id":"account","type":"user","name":"mesh-mail-send"}]}`)) + if err != nil { + t.Fatalf("a manifest with own=%q was refused: %v", own, err) + } + return m +} + +// 1170/AC1: a family owned by the module's own account lands at 0600, owned by that account — and a module's +// other own secrets (no per-secret owner declared) keep secrets-owner, unchanged. +func TestAnOwnSecretsOwnerIsGivenToItsFiles(t *testing.T) { + m := mailSenderManifest(t, "mesh-mail-send") + m.OwnSecrets["smtp-relay-key"] = OwnSecret{Path: "/var/lib/mesh-mail/smtp-relay-key", Taken: TakenAtStart} + r := Resolution{Node: "novox", Modules: []Manifest{m}} + out, err := r.Declaration(Rendering{ + Needed: map[string]map[string]string{"mesh-mail": {"mail-credential-smtp": "sealed-credential", "smtp-relay-key": "sealed-relay"}}, + }) + if err != nil { + t.Fatalf("composing the declaration: %v", err) + } + var family, plain map[string]any + for _, res := range out { + if res["type"] != "file" { + continue + } + switch res["path"] { + case "/var/lib/mesh-mail/mail-credential-smtp.secret": + family = res + case "/var/lib/mesh-mail/smtp-relay-key": + plain = res + } + } + if family == nil { + t.Fatalf("the family member was not composed at all: %v", out) + } + if got := family["owner"]; got != "mesh-mail-send" { + t.Fatalf("the family member's owner: %v; want mesh-mail-send", got) + } + // review round 1, finding 7: asserted, not merely logged. ownedBy never sets "mode", so a sealed secret + // file carries none here either — the host applies 0600 to every one regardless (novox/hq ADR 0086) — + // and the owner clause must not have added one of its own that could override that. + if _, has := family["mode"]; has { + t.Fatalf("the family member carries its own mode %v; the owner clause must not set one, so the "+ + "host's 0600 for every sealed secret file stands", family["mode"]) + } + if plain == nil { + t.Fatalf("the module's other own secret was not composed at all: %v", out) + } + if got := plain["owner"]; got != "mesh-mail" { + t.Fatalf("a secret with no per-secret owner kept %v; want the module's secrets-owner %q", got, "mesh-mail") + } +} + +// 1170/AC2: an owner naming an account the module does not make, root, or the operator account's placeholder is +// refused, naming the secret and the account. +func TestAnOwnSecretsOwnerIsRefusedWhenItIsNotAnAccountTheModuleMakes(t *testing.T) { + for name, c := range map[string]struct{ owner, says string }{ + "root": {"root", "root"}, + "the operator account": {"${machine:account}", "operator account"}, + "an account not declared": {"not-made", "not-made"}, + } { + _, err := ParseManifest([]byte(`{"module":"mesh-mail","version":"1","own-secrets":{` + + `"mail-credential-*":{"path":"/var/lib/mesh-mail/mail-credential-*.secret","issued-by":"outside","owner":"` + + c.owner + `"}},"resources":[{"id":"account","type":"user","name":"mesh-mail-send"}]}`)) + if err == nil || !strings.Contains(err.Error(), "mail-credential-*") || !strings.Contains(err.Error(), c.says) { + t.Errorf("%s: %v; want a refusal naming the secret and %q", name, err, c.says) + } + } +} + +// 1170/AC3: a manifest with no owner on any own secret composes exactly as it did before this clause existed: +// every secret file belongs to the module's secrets-owner (or nothing, when that is absent too). +func TestNoOwnerOnAnySecretComposesAsMainDoes(t *testing.T) { + m := mailSenderManifest(t, "") + r := Resolution{Node: "novox", Modules: []Manifest{m}} + out, err := r.Declaration(Rendering{ + Needed: map[string]map[string]string{"mesh-mail": {"mail-credential-smtp": "sealed-credential"}}, + }) + if err != nil { + t.Fatalf("composing the declaration: %v", err) + } + var found bool + for _, res := range out { + if res["type"] == "file" && res["path"] == "/var/lib/mesh-mail/mail-credential-smtp.secret" { + found = true + if got := res["owner"]; got != "mesh-mail" { + t.Fatalf("with no per-secret owner declared, the file's owner: %v; want secrets-owner %q", got, "mesh-mail") + } + } + } + if !found { + t.Fatalf("the family member was not composed at all: %v", out) + } +} diff --git a/internal/catalogue/process_bus.go b/internal/catalogue/process_bus.go new file mode 100644 index 00000000..445fc68a --- /dev/null +++ b/internal/catalogue/process_bus.go @@ -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 . or a role's + // seat:. (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 "+ + "., or a role's seat:. (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: ., a role's +// seat:., 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) +} diff --git a/internal/catalogue/process_bus_test.go b/internal/catalogue/process_bus_test.go new file mode 100644 index 00000000..5551ec17 --- /dev/null +++ b/internal/catalogue/process_bus_test.go @@ -0,0 +1,191 @@ +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 += `}` + // The stream itself is plain here, not retention work-queue: 1179 checks only that a process names a + // stream the module owns, never its retention — that rule is 1178's (WorkQueueStreamProblems), and is + // tested against its own fixture there (work_queue_consumer_test.go) so this one stays accepted whether + // or not a process currently names "mail-sends" in bus.queues. + 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}],"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 +} diff --git a/internal/catalogue/seats_declared.go b/internal/catalogue/seats_declared.go index d53d3e04..2a9ad3fe 100644 --- a/internal/catalogue/seats_declared.go +++ b/internal/catalogue/seats_declared.go @@ -280,6 +280,8 @@ func CatalogueProblems(shelf Shelf) []string { problems = append(problems, StreamNameProblems(shelf)...) // A module that hears a restricted stream runs as an account of its own (novox/hq ADR 0376 §2). problems = append(problems, RestrictedHearingProblems(shelf)...) + // A work-queue stream is never asked for by another module's consumes (novox/hq ADR 0436 §6). + problems = append(problems, WorkQueueConsumedElsewhereProblems(shelf)...) // Who declares what, and who declared it first. declaredBy := map[string]string{} diff --git a/internal/catalogue/stream_work_queue_test.go b/internal/catalogue/stream_work_queue_test.go new file mode 100644 index 00000000..9f8ccf64 --- /dev/null +++ b/internal/catalogue/stream_work_queue_test.go @@ -0,0 +1,71 @@ +package catalogue + +import ( + "strings" + "testing" +) + +// A module stream may declare retention: work-queue (novox/hq ADR 0436 §6): a message is removed once a +// consumer acknowledges it, the bus refusing a new one rather than dropping what it holds. The fixture is +// mesh-mail's future `mail-sends` stream (ADR 0436 §7, §8), copied from the module.json shape at +// /home/jochen/projects/novox/mesh-catalog/modules/mesh-mail/module.json and extended with retention, +// max-deliveries and duplicates as the ADR's clause 6 describes. +const mailSendsStream = `[{"name":"mail-sends","events":["mail.send.approved"],"max-age-days":1,"max-mib":64,` + + `"retention":"work-queue","max-deliveries":5,"duplicates":24}]` + +// 1177/AC1: a well-formed work-queue stream is accepted and read back whole. Named by one process in +// bus.queues (novox/hq ADR 0436 §6, checked by 1178's WorkQueueStreamProblems), so this stays accepted once +// that rule exists. +func TestAWorkQueueStreamIsDeclaredAndAccepted(t *testing.T) { + m, err := ParseManifest([]byte(`{"module":"mesh-mail","version":"1","streams":` + mailSendsStream + + `,"resources":[{"id":"account","type":"user","name":"mesh-mail-send"},` + + `{"id":"send","type":"process","run":["./mesh-mail","send"],"user":"mesh-mail-send",` + + `"bus":{"queues":["mail-sends"]}}]}`)) + if err != nil { + t.Fatalf("a well-formed work-queue stream was refused: %v", err) + } + s := m.Streams[0] + if s.Retention != StreamRetentionWorkQueue || s.MaxDeliveries != 5 || s.Duplicates != 24 || + s.Days() != 1 || s.Cap() != 64 { + t.Fatalf("the stream read as %+v", s) + } +} + +// 1177/AC2: a stream whose work-queue values are out of range, contradictory, or carries a grant is refused +// naming the stream and the rule. +func TestAWorkQueueStreamsValuesAreBounded(t *testing.T) { + for name, c := range map[string]struct{ streams, says string }{ + "max-deliveries 0": {`[{"name":"mail-sends","events":["mail.send.approved"],"retention":"work-queue",` + + `"max-deliveries":0,"duplicates":24}]`, "between 1 and 10"}, + "max-deliveries 11": {`[{"name":"mail-sends","events":["mail.send.approved"],"retention":"work-queue",` + + `"max-deliveries":11,"duplicates":24}]`, "between 1 and 10"}, + "duplicates longer than its maximum age": {`[{"name":"mail-sends","events":["mail.send.approved"],` + + `"max-age-days":1,"retention":"work-queue","max-deliveries":5,"duplicates":25}]`, "longer than its maximum age"}, + "an unknown retention": {`[{"name":"mail-sends","events":["mail.send.approved"],"retention":"sometimes"}]`, + "retention"}, + "max-deliveries without retention work-queue": {`[{"name":"mail-sends","events":["mail.send.approved"],` + + `"max-deliveries":5}]`, "without retention"}, + "duplicates without retention work-queue": {`[{"name":"mail-sends","events":["mail.send.approved"],` + + `"duplicates":24}]`, "without retention"}, + "a work-queue stream carrying a grant": {`[{"name":"mail-sends","events":["mail.send.approved"],` + + `"retention":"work-queue","max-deliveries":5,` + + `"grant":{"config":"grants","events":"mail-sends.{key}.**"}}]`, "carrying a grant"}, + } { + _, err := ParseManifest([]byte(mailManifest(t, c.streams, ""))) + if err == nil || !strings.Contains(err.Error(), "mail-sends") || !strings.Contains(err.Error(), c.says) { + t.Errorf("%s: %v; want a refusal naming the stream and %q", name, err, c.says) + } + } +} + +// 1177/AC3: a stream with no retention composes as main does — limits retention, dropping its oldest — proven +// at the broker, where the bus's retention policy is decided (internal/broker/modstreams_test.go). +func TestAStreamWithNoRetentionReadsAsLimits(t *testing.T) { + m, err := ParseManifest([]byte(mailManifest(t, mailStream, ""))) + if err != nil { + t.Fatal(err) + } + if got := m.Streams[0].Retention; got != "" { + t.Fatalf("a stream declaring no retention read %q", got) + } +} diff --git a/internal/catalogue/streams.go b/internal/catalogue/streams.go index 012178b6..15c6c2b0 100644 --- a/internal/catalogue/streams.go +++ b/internal/catalogue/streams.go @@ -30,6 +30,18 @@ const ( // GrantKey is the name in a grant's pattern that each key of the config value fills. const GrantKey = "{key}" +// StreamRetentionWorkQueue is a stream's declared retention (novox/hq ADR 0436 §6): a message is removed +// once a consumer acknowledges it, and the bus refuses a new one rather than dropping what it holds. Empty +// is the mesh's own limits retention — kept until it ages or the stream fills, dropping its oldest. +const StreamRetentionWorkQueue = "work-queue" + +// StreamLeastDeliveries and StreamMostDeliveries bound a work-queue stream's max-deliveries (novox/hq ADR +// 0436 §6): how many times a message is handed over before it is given up as a dead letter. +const ( + StreamLeastDeliveries = 1 + StreamMostDeliveries = 10 +) + // ReservedStreamPrefixes are the starts of the stream names the mesh derives, which no module's stream // may take: a seat's work queue, a log, a bucket, an object store, and the dead letters. var ReservedStreamPrefixes = []string{"SEAT_", "LOG_", "KV_", "OBJ_", "DEAD_"} @@ -48,6 +60,15 @@ type StreamDeclaration struct { MaxMiB int `json:"max-mib,omitempty"` // Grant restricts it, when said. Grant *StreamGrant `json:"grant,omitempty"` + // Retention is how the stream decides what to keep when it fills (novox/hq ADR 0436 §6): empty for the + // mesh's own limits retention (dropping its oldest), or StreamRetentionWorkQueue. + Retention string `json:"retention,omitempty"` + // MaxDeliveries bounds how many times a work-queue message is handed over, 1 to 10; refused without + // retention work-queue (novox/hq ADR 0436 §6). + MaxDeliveries int `json:"max-deliveries,omitempty"` + // Duplicates is a work-queue stream's duplicate window in hours, at most its maximum age; refused + // without retention work-queue (novox/hq ADR 0436 §6). + Duplicates int `json:"duplicates,omitempty"` } // StreamGrant says which trusted config value of the module grants a consumer a part of the stream, and @@ -67,7 +88,8 @@ func (s *StreamDeclaration) UnmarshalJSON(raw []byte) error { dec := json.NewDecoder(bytes.NewReader(raw)) dec.DisallowUnknownFields() if err := dec.Decode(&p); err != nil { - return fmt.Errorf("a stream is {name, events, max-age-days, max-mib, grant}: %w", typedUnknown(err)) + return fmt.Errorf("a stream is {name, events, max-age-days, max-mib, grant, retention, max-deliveries, "+ + "duplicates}: %w", typedUnknown(err)) } *s = StreamDeclaration(p) return nil @@ -163,6 +185,44 @@ func StreamProblems(m Manifest) []string { problems = append(problems, fmt.Sprintf("%s caps stream %q at %d MiB; a stream holds between %d and %d "+ "(novox/hq ADR 0363)", m.Module, s.Name, s.MaxMiB, StreamLeastMiB, StreamMostMiB)) } + // A stream may declare retention: work-queue (novox/hq ADR 0436 §6): a message removed once a + // consumer acknowledges it, the bus refusing a new one when full rather than dropping its oldest, with + // max-deliveries (1 to 10) and a duplicate window (duplicates, hours, at most its maximum age) -- both + // meaningless, and refused, without it. A work-queue stream is read by its one consumer, never granted. + switch s.Retention { + case "", StreamRetentionWorkQueue: + default: + problems = append(problems, fmt.Sprintf( + "%s's stream %q declares retention %q; a stream's retention is %q or nothing (novox/hq ADR 0436 §6)", + m.Module, s.Name, s.Retention, StreamRetentionWorkQueue)) + } + switch { + case s.Retention == StreamRetentionWorkQueue && + (s.MaxDeliveries < StreamLeastDeliveries || s.MaxDeliveries > StreamMostDeliveries): + problems = append(problems, fmt.Sprintf( + "%s's stream %q declares max-deliveries %d; a work-queue stream's max-deliveries is between %d "+ + "and %d (novox/hq ADR 0436 §6)", m.Module, s.Name, s.MaxDeliveries, StreamLeastDeliveries, + StreamMostDeliveries)) + case s.Retention != StreamRetentionWorkQueue && s.MaxDeliveries != 0: + problems = append(problems, fmt.Sprintf( + "%s's stream %q declares max-deliveries %d without retention %q (novox/hq ADR 0436 §6)", + m.Module, s.Name, s.MaxDeliveries, StreamRetentionWorkQueue)) + } + switch { + case s.Duplicates != 0 && s.Retention != StreamRetentionWorkQueue: + problems = append(problems, fmt.Sprintf( + "%s's stream %q declares duplicates %d hours without retention %q (novox/hq ADR 0436 §6)", + m.Module, s.Name, s.Duplicates, StreamRetentionWorkQueue)) + case s.Duplicates != 0 && s.Duplicates > s.Days()*24: + problems = append(problems, fmt.Sprintf( + "%s's stream %q keeps duplicates %d hours, longer than its maximum age of %d days (novox/hq "+ + "ADR 0436 §6)", m.Module, s.Name, s.Duplicates, s.Days())) + } + if s.Retention == StreamRetentionWorkQueue && s.Grant != nil { + problems = append(problems, fmt.Sprintf( + "%s's stream %q declares retention %q carrying a grant; a work-queue stream is read by its "+ + "module's one consumer, never granted (novox/hq ADR 0436 §6)", m.Module, s.Name, s.Retention)) + } if len(s.Events) == 0 { problems = append(problems, fmt.Sprintf("%s declares stream %q with no events", m.Module, s.Name)) } @@ -291,6 +351,64 @@ func patternsOverlap(a, b string) bool { } } +// WorkQueueStreamProblems refuses a work-queue stream (novox/hq ADR 0436 §6) that is not read by exactly +// one process of its own module: named by none, it queues forever unconsumed; named by two, each would take +// half of what the other needed, silently. +func WorkQueueStreamProblems(m Manifest) []string { + var problems []string + named := map[string]int{} + for _, b := range m.ProcessBuses() { + for _, q := range b.Queues { + named[q]++ + } + } + for _, s := range m.Streams { + if s.Retention != StreamRetentionWorkQueue { + continue + } + switch named[s.Name] { + case 0: + problems = append(problems, fmt.Sprintf( + "%s's work-queue stream %q: no process names it in bus.queues; a work-queue stream is read by "+ + "exactly one process (novox/hq ADR 0436 §6)", m.Module, s.Name)) + case 1: + default: + problems = append(problems, fmt.Sprintf( + "%s's work-queue stream %q: %d processes name it in bus.queues; a work-queue stream is read by "+ + "exactly one (novox/hq ADR 0436 §6)", m.Module, s.Name, named[s.Name])) + } + } + return problems +} + +// WorkQueueConsumedElsewhereProblems refuses another module's `consumes` naming a work-queue stream's events +// (novox/hq ADR 0436 §6): read by its module's one process alone, never broadcast to a consumer that only +// subscribes. Modelled on RestrictedHearingProblems. +func WorkQueueConsumedElsewhereProblems(shelf Shelf) []string { + var problems []string + for _, module := range shelfOrder(shelf) { + m := shelf[module] + for _, c := range m.Consumes { + emitter, rest, named := strings.Cut(c, ".") + provider, known := shelf[emitter] + if !named || emitter == OneName || emitter == TheRest || !known || emitter == module { + continue + } + hearing, _ := provider.StreamsHearing(rest) + for _, s := range hearing { + if s.Retention != StreamRetentionWorkQueue { + continue + } + problems = append(problems, fmt.Sprintf( + "%s consumes %s, which %s's work-queue stream %s carries; a work-queue stream is read by "+ + "its module's one process alone, never by another module's consumes (novox/hq ADR 0436 "+ + "§6)", module, c, emitter, s.BusName())) + } + } + } + return problems +} + // StreamNameProblems are the rules about stream names no single manifest can be judged against: two // modules declaring one name. func StreamNameProblems(shelf Shelf) []string { diff --git a/internal/catalogue/work_queue_consumer_test.go b/internal/catalogue/work_queue_consumer_test.go new file mode 100644 index 00000000..4f98b6f1 --- /dev/null +++ b/internal/catalogue/work_queue_consumer_test.go @@ -0,0 +1,75 @@ +package catalogue + +import ( + "strings" + "testing" +) + +// A work-queue stream is read by exactly one process of its own module (novox/hq ADR 0436 §6): named once +// in a process's bus.queues, and never asked for by another module's consumes. + +// mailSenderManifestWithQueue is the mesh-mail fixture (ADR 0436 §7, §8) with its process's bus.queues set +// as given, so each 1178/AC2 case can vary just that. +func mailSenderManifestWithQueue(queues string) Manifest { + return Manifest{ + Module: "mesh-mail", Version: "1", Emits: []string{"mail.send.sent", "mail.send.failed"}, + State: []StateDeclaration{{Name: "sends"}}, + Streams: []StreamDeclaration{{Name: "mail-sends", Events: []string{"mail.send.approved"}, + Retention: StreamRetentionWorkQueue, MaxDeliveries: 5}}, + Resources: []map[string]any{ + {"id": "account", "type": "user", "name": "mesh-mail-send"}, + {"id": "send", "type": "process", "run": []any{"./mesh-mail", "send"}, "user": "mesh-mail-send", + "bus": map[string]any{"emits": []any{"mail.send.sent", "mail.send.failed"}, "queues": queueList(queues)}}, + }, + } +} + +func queueList(q string) []any { + if q == "" { + return nil + } + out := []any{} + for _, s := range strings.Split(q, ",") { + out = append(out, s) + } + return out +} + +// 1178/AC2 (named): a work-queue stream no process names is refused. +func TestAWorkQueueStreamNoProcessNamesIsRefused(t *testing.T) { + m := mailSenderManifestWithQueue("") + got := strings.Join(WorkQueueStreamProblems(m), "; ") + if !strings.Contains(got, "mail-sends") || !strings.Contains(got, "no process names it") { + t.Fatalf("an unnamed work-queue stream: %q", got) + } +} + +// 1178/AC2 (named twice): a work-queue stream two processes name is refused. +func TestAWorkQueueStreamTwoProcessesNameIsRefused(t *testing.T) { + m := mailSenderManifestWithQueue("mail-sends") + m.Resources = append(m.Resources, map[string]any{"id": "resend", "type": "process", + "run": []any{"./mesh-mail", "send"}, "user": "mesh-mail-send", + "bus": map[string]any{"emits": []any{"mail.send.sent"}, "queues": []any{"mail-sends"}}}) + got := strings.Join(WorkQueueStreamProblems(m), "; ") + if !strings.Contains(got, "mail-sends") || !strings.Contains(got, "2 processes name") { + t.Fatalf("a work-queue stream two processes name: %q", got) + } +} + +// 1178/AC2 (named once): a work-queue stream exactly one process names is accepted. +func TestAWorkQueueStreamOneProcessNamesIsAccepted(t *testing.T) { + m := mailSenderManifestWithQueue("mail-sends") + if got := WorkQueueStreamProblems(m); len(got) != 0 { + t.Fatalf("a work-queue stream one process names was refused: %v", got) + } +} + +// 1178/AC2 (another module's consumes): a work-queue stream another module's consumes asks for is refused. +func TestAWorkQueueStreamAnotherModulesConsumesIsRefused(t *testing.T) { + sender := mailSenderManifestWithQueue("mail-sends") + reader := Manifest{Module: "reader", Version: "1", Consumes: []string{"mesh-mail.mail.send.approved"}} + got := strings.Join(WorkQueueConsumedElsewhereProblems(Shelf{"mesh-mail": sender, "reader": reader}), "; ") + if !strings.Contains(got, "reader") || !strings.Contains(got, "MAIL_SENDS") { + t.Fatalf("another module's consumes of a work-queue stream: %q", got) + } +} diff --git a/internal/inventory/busrecords.go b/internal/inventory/busrecords.go index e8b78364..09fd6215 100644 --- a/internal/inventory/busrecords.go +++ b/internal/inventory/busrecords.go @@ -187,6 +187,12 @@ func declaredFor(m catalogue.Manifest, seats map[string]catalogue.SeatDeclaratio Logs: logsOf(m), // And the streams it owns (novox/hq ADR 0363). Streams: streamsOf(m), + // And its own work-queue streams, each with the one process its bus.queues names as the + // consumer (novox/hq ADR 0436 §1, §2, §6). + Queues: queueOwnersOf(m), + // And every process of its own that declares bus, each granted a bus account of its own (novox/hq + // ADR 0436 §1, §2). + Processes: ProcessesOf(m), // And the tools its health asks (novox/hq ADR 0240): the machine's node-engine is granted them. Checks: catalogue.HealthChecks(m), // And whether it runs as an account of its own (novox/hq ADR 0259 §8). @@ -422,7 +428,67 @@ func streamsOf(m catalogue.Manifest) []broker.ModuleStream { events = append(events, broker.WirePattern(e)) } out = append(out, broker.ModuleStream{Module: m.Module, Name: s.BusName(), Events: events, - MaxAgeDays: s.Days(), MaxMiB: s.Cap(), Restricted: s.Grant != nil}) + MaxAgeDays: s.Days(), MaxMiB: s.Cap(), Restricted: s.Grant != nil, + WorkQueue: s.Retention == catalogue.StreamRetentionWorkQueue, Duplicates: s.Duplicates}) + } + return out +} + +// queueOwnersOf is every work-queue stream of a module's own that one of its processes names in bus.queues, +// with that process's id and the stream's declared max-deliveries (novox/hq ADR 0436 §1, §2, §6). In +// resource order, so composition is repeatable; a queue naming a stream the module does not own, or one two +// processes name, is caught at registration (catalogue.WorkQueueStreamProblems) and never reaches here. +func queueOwnersOf(m catalogue.Manifest) []broker.QueueOwner { + buses := m.ProcessBuses() + if len(buses) == 0 { + return nil + } + byName := map[string]catalogue.StreamDeclaration{} + for _, s := range m.Streams { + byName[s.Name] = s + } + var out []broker.QueueOwner + for _, r := range m.Resources { + id := fmt.Sprint(r["id"]) + b, declared := buses[id] + if !declared { + continue + } + for _, q := range b.Queues { + s, known := byName[q] + if !known || s.Retention != catalogue.StreamRetentionWorkQueue { + continue + } + out = append(out, broker.QueueOwner{Process: id, Stream: s.BusName(), MaxDeliver: s.MaxDeliveries}) + } + } + return out +} + +// ProcessesOf is every process of a module that declares bus, as the broker composes its own account from +// (novox/hq ADR 0436 §1, §2): its own emits, its own work-queue consumers (queueOwnersOf, filtered to each +// process's own), its own state, whether it reads its module's own asks, and the tools and seat verbs it +// calls. +func ProcessesOf(m catalogue.Manifest) []broker.ProcessPrincipal { + buses := m.ProcessBuses() + if len(buses) == 0 { + return nil + } + queues := queueOwnersOf(m) + var out []broker.ProcessPrincipal + for _, r := range m.Resources { + id := fmt.Sprint(r["id"]) + b, declared := buses[id] + if !declared { + continue + } + pr := broker.ProcessPrincipal{Process: id, Emits: b.Emits, State: b.State, Asks: b.Asks, Invokes: b.Invokes} + for _, q := range queues { + if q.Process == id { + pr.Queues = append(pr.Queues, q) + } + } + out = append(out, pr) } return out } diff --git a/internal/inventory/bususers.go b/internal/inventory/bususers.go index 6a954909..388e6aaf 100644 --- a/internal/inventory/bususers.go +++ b/internal/inventory/bususers.go @@ -68,6 +68,9 @@ const ( // BusProvisioner is a machine's provisioner client (novox/hq ADR 0349 clause 11, design 57 P3): an account of // its own, named for the machine. BusProvisioner = "provisioner" + // BusProcess is one process of a module that declares bus (novox/hq ADR 0436 §1, §2): an account of its + // own, named for the machine, the module and the process. + BusProcess = "process" ) // MintBusPassword makes a bus password and records its hash under a username, replacing whatever was diff --git a/internal/inventory/streams_test.go b/internal/inventory/streams_test.go index dbddf979..4a4cd0fb 100644 --- a/internal/inventory/streams_test.go +++ b/internal/inventory/streams_test.go @@ -43,3 +43,47 @@ func TestWhatAModuleHearsIsComposedOntoTheStreamsSubjects(t *testing.T) { t.Fatalf("the provider's stream: %+v", got) } } + +// 1177/AC1: a catalogue stream declaring retention work-queue is carried into the broker's ModuleStream as +// WorkQueue, with its duplicate window. +func TestAWorkQueueStreamIsCarriedIntoTheBrokersModuleStream(t *testing.T) { + sender := catalogue.Manifest{Module: "mesh-mail", Streams: []catalogue.StreamDeclaration{{Name: "mail-sends", + Events: []string{"mail.send.approved"}, MaxAgeDays: 1, MaxMiB: 64, + Retention: catalogue.StreamRetentionWorkQueue, MaxDeliveries: 5, Duplicates: 24}}} + got := streamsOf(sender) + if len(got) != 1 || got[0].Name != "MAIL_SENDS" || !got[0].WorkQueue || got[0].Duplicates != 24 { + t.Fatalf("the sender's stream: %+v", got) + } +} + +// 1177/AC3: a catalogue stream with no retention is carried into the broker as not a work queue. +func TestANonWorkQueueStreamIsCarriedAsNotAWorkQueue(t *testing.T) { + plain := catalogue.Manifest{Module: "feed", Streams: []catalogue.StreamDeclaration{{Name: "items", + Events: []string{"item.**"}}}} + got := streamsOf(plain) + if len(got) != 1 || got[0].WorkQueue || got[0].Duplicates != 0 { + t.Fatalf("the plain stream: %+v", got) + } +} + +// 1178/AC1: the process a work-queue stream's bus.queues names is read from the manifest, by its resource +// id, with the stream's own max-deliveries — what the broker composes the one durable consumer from. +func TestTheQueueOwnersOfAWorkQueueStreamAreReadFromItsProcess(t *testing.T) { + sender := catalogue.Manifest{Module: "mesh-mail", Emits: []string{"mail.send.sent", "mail.send.failed"}, + Streams: []catalogue.StreamDeclaration{{Name: "mail-sends", Events: []string{"mail.send.approved"}, + Retention: catalogue.StreamRetentionWorkQueue, MaxDeliveries: 5}}, + Resources: []map[string]any{ + {"id": "account", "type": "user", "name": "mesh-mail-send"}, + {"id": "send", "type": "process", "run": []any{"./mesh-mail", "send"}, "user": "mesh-mail-send", + "bus": map[string]any{"emits": []any{"mail.send.sent", "mail.send.failed"}, + "queues": []any{"mail-sends"}}}, + }} + got := queueOwnersOf(sender) + if len(got) != 1 || got[0].Process != "send" || got[0].Stream != "MAIL_SENDS" || got[0].MaxDeliver != 5 { + t.Fatalf("the sender's queue owners: %+v", got) + } + // And a module with no process declaring bus has none. + if got := queueOwnersOf(catalogue.Manifest{Module: "feed"}); got != nil { + t.Fatalf("a module with no process bus has queue owners: %+v", got) + } +} diff --git a/internal/link/deadletters.go b/internal/link/deadletters.go index 94b442cb..94de7d22 100644 --- a/internal/link/deadletters.go +++ b/internal/link/deadletters.go @@ -45,6 +45,10 @@ const ( // DeadHeaderAgesOut when its stream's maximum age runs out, counted from when it was first published. DeadHeaderRestricted = "Mesh-Dead-Restricted" DeadHeaderAgesOut = "Mesh-Dead-Ages-Out" + // DeadHeaderWorkQueue says the message came from a work-queue stream (novox/hq ADR 0436 §6): aged and + // shown as a restricted stream's are (it sets DeadHeaderRestricted too), but never delivered again — + // only dropped. + DeadHeaderWorkQueue = "Mesh-Dead-Work-Queue" // AgainHeader marks a message delivered again, with the kept copy's id it came from. AgainHeader = "Mesh-Delivered-Again" ) @@ -74,6 +78,10 @@ type DeadLetter struct { Restricted bool `json:"restricted,omitempty"` AgesOut time.Time `json:"ages_out,omitzero"` Withheld string `json:"withheld,omitempty"` + // WorkQueue says it came from a work-queue stream (novox/hq ADR 0436 §6): shown and aged as a restricted + // stream's are (Restricted is also set), but AgainTo refuses it unconditionally — only dropped, never + // delivered again. + WorkQueue bool `json:"work-queue,omitempty"` // Body and Headers are the message itself, given only for one dead letter asked by its id, and never for // a restricted one; a header with several values keeps them all. Body string `json:"body,omitempty"` @@ -91,17 +99,27 @@ const leastLeftToDeliver = time.Minute const withheldWords = "its body and its own headers are not shown: it came from a restricted stream, which only " + "the modules its grant names may read (novox/hq ADR 0376)" -// sourceAge is whether a stream is restricted (novox/hq ADR 0363 §4), and its maximum age: read from the stream -// itself, where the controller says both. A stream that no longer exists is neither. -func sourceAge(js nats.JetStreamContext, stream string) (bool, time.Duration, error) { +// sourceAge is whether a stream is restricted (novox/hq ADR 0363 §4; a work-queue stream is restricted too, +// novox/hq ADR 0436 §6, aged and hidden the same way), whether it is a module stream declared retention: +// work-queue, and its maximum age: read from the stream itself, where the controller says all three. A +// stream that no longer exists is neither. +// +// **The module declaration, never the bus's work-queue retention alone** (novox/hq ADR 0436 §6, found by +// mesh-controller.try on d8249dad): SEAT_* and other mesh-internal streams are JetStream work queues too — +// exactly one worker takes each ask (ADR 0264 §4) — and their dead letters are still delivered again. Only +// a module stream the catalogue declared retention: work-queue says so in its own metadata +// (broker.WorkQueueMetadata, set by ModuleStream.Stream() alone); checking the bus's retention policy +// caught every seat's queue too and refused every ask ever given up on delivered again. +func sourceAge(js nats.JetStreamContext, stream string) (restricted, workQueue bool, maxAge time.Duration, err error) { info, err := js.StreamInfo(stream) if errors.Is(err, nats.ErrStreamNotFound) { - return false, 0, nil + return false, false, 0, nil } if err != nil { - return false, 0, fmt.Errorf("whether %s is restricted cannot be read: %w", stream, err) + return false, false, 0, fmt.Errorf("whether %s is restricted cannot be read: %w", stream, err) } - return info.Config.Metadata[broker.RestrictedMetadata] == "true", info.Config.MaxAge, nil + return info.Config.Metadata[broker.RestrictedMetadata] == "true", + info.Config.Metadata[broker.WorkQueueMetadata] == "true", info.Config.MaxAge, nil } // ttlOf is a time left as a message's time to live, in whole seconds and never more than what is left. @@ -135,13 +153,16 @@ func KeepDeadLetter(js nats.JetStreamContext, notice []byte) (DeadLetter, error) // **A restricted stream's message ages out with its stream** (novox/hq ADR 0376 §3): kept with a time to live // of what is left of the stream's maximum age, counted from when it was first published; one already past it // is kept as a record without its body. Every other dead letter keeps no age, as before (ADR 0264). - restricted, maxAge, err := sourceAge(js, a.Stream) + restricted, workQueue, maxAge, err := sourceAge(js, a.Stream) if err != nil { return DeadLetter{}, err } if restricted { kept.Header.Set(DeadHeaderRestricted, "true") } + if workQueue { + kept.Header.Set(DeadHeaderWorkQueue, "true") + } raw, err := js.GetMsg(a.Stream, a.StreamSeq) agesOut := time.Time{} if err == nil && restricted && maxAge > 0 { @@ -205,6 +226,7 @@ func deadLetterOf(id uint64, subject string, h nats.Header, data []byte, whole b d.GaveUp, _ = time.Parse(time.RFC3339Nano, h.Get(DeadHeaderGaveUp)) d.Published, _ = time.Parse(time.RFC3339Nano, h.Get(DeadHeaderPublished)) d.Restricted = h.Get(DeadHeaderRestricted) == "true" + d.WorkQueue = h.Get(DeadHeaderWorkQueue) == "true" d.AgesOut, _ = time.Parse(time.RFC3339Nano, h.Get(DeadHeaderAgesOut)) if whole { d.Body = string(data) @@ -349,6 +371,13 @@ func AgainTo(js nats.JetStreamContext, d DeadLetter) (string, error) { case d.Subject == "": return "", fmt.Errorf("dead letter %d does not say the subject it was published on, so it cannot be "+ "delivered again. Drop it", d.ID) + case d.WorkQueue: + // A work-queue stream's one consumer takes its messages in order (novox/hq ADR 0436 clause 6): + // delivering a given-up one back in would let it jump ahead of whatever the stream already holds, so + // unlike a module's ordinary stream (ADR 0363 §7) it is never delivered again, however much of its + // age is left. + return "", fmt.Errorf("dead letter %d is from %s, a work queue: it is only dropped, never delivered "+ + "again (novox/hq ADR 0436 clause 6). Drop it", d.ID, d.Stream) case broker.TheControllersAsk(d.Stream, d.Subject): // The seat's worker takes it, and only while it is on the bus: without one the queue would keep // the ask for a holder that may never come, and it would be let go as delivered meanwhile. @@ -593,7 +622,7 @@ func AgingDeadLetters(js nats.JetStreamContext, held map[string]int) (map[string stream, consumer, _ := strings.Cut(key, ".") is, known := restricted[stream] if !known { - r, _, err := sourceAge(js, stream) + r, _, _, err := sourceAge(js, stream) if err != nil { return nil, err } diff --git a/internal/link/deadletters_work_queue_test.go b/internal/link/deadletters_work_queue_test.go new file mode 100644 index 00000000..2a8d6946 --- /dev/null +++ b/internal/link/deadletters_work_queue_test.go @@ -0,0 +1,88 @@ +package link + +import ( + "strings" + "testing" + "time" + + "github.com/nats-io/nats.go" + + "github.com/novox/mesh-controller/internal/broker" +) + +// A work-queue stream's dead letters are kept and shown as a restricted stream's are — aged with the +// stream, never with their body or headers — and can only be dropped, never delivered again (novox/hq ADR +// 0436 §6, changing ADR 0363 §7 for such a stream). + +// review round 1, finding 6 (1178/AC3): a work-queue stream really is restricted on the bus — its own +// Stream() says so, and that is what EnsureStream's config sets the metadata from (jetstream.go's +// streamConfigFor) — not merely a header this test set by hand, which would still pass with +// modstreams.go's `out.Restricted = true` removed. +func TestAWorkQueueStreamIsRestrictedOnTheBus(t *testing.T) { + mailSends := broker.ModuleStream{Module: "mesh-mail", Name: "MAIL_SENDS", Events: []string{"mail.send.approved"}, + MaxAgeDays: 1, MaxMiB: 8, WorkQueue: true, Duplicates: 1} + stream := mailSends.Stream() + if !stream.Restricted { + t.Fatalf("a work-queue stream's own Stream() is not restricted: %+v", stream) + } + cfg := broker.StreamConfigFor(stream) + if cfg.Metadata[broker.RestrictedMetadata] != "true" { + t.Fatalf("the server config EnsureStream would send does not carry the restricted metadata: %+v", cfg.Metadata) + } +} + +// mesh-controller.try fault A, d8249dad: a seat's own work queue (SEAT_*, novox/hq ADR 0190, ADR 0264 §4) is +// a JetStream work queue at the bus level too — exactly one worker takes each ask — but it is never a module +// stream the catalogue declared retention: work-queue (ADR 0436 §6), so it carries none of that clause's +// metadata and its dead letters are still delivered again (ADR 0264 §4), unlike mail-sends'. +func TestASeatsWorkQueueIsNotADR0436WorkQueue(t *testing.T) { + seatQueue := broker.Stream{Name: "SEAT_NODE_BUILD_AGENT", Retention: broker.RetentionWorkQueue, DiscardNew: true} + cfg := broker.StreamConfigFor(seatQueue) + if _, has := cfg.Metadata[broker.WorkQueueMetadata]; has { + t.Fatalf("a seat's own work queue carries ADR 0436 §6's metadata: %+v", cfg.Metadata) + } + if _, has := cfg.Metadata[broker.RestrictedMetadata]; has { + t.Fatalf("a seat's own work queue is restricted: %+v", cfg.Metadata) + } + // sourceAge (deadletters.go) reads exactly these two keys off a real stream's config; neither set means + // KeepDeadLetter marks a seat's dead letter WorkQueue: false, so AgainTo never reaches clause 6's + // drop-only case for it (proven directly in TestAWorkQueueDeadLetterIsNeverDeliveredAgain) and a seat's + // ask is still delivered again (ADR 0264 §4), as it was before this release. +} + +// 1178/AC3: a dead letter's headers say it came from a work-queue stream, and reading it back carries that +// (and the restricted flag its aging and hiding already key off, deadLetterOf being the one place both are +// read: see DeadLetter.Restricted's doc). +func TestAWorkQueueDeadLettersHeaderIsReadBack(t *testing.T) { + agesOut := time.Now().Add(time.Hour).UTC() + h := nats.Header{} + h.Set(DeadHeaderStream, "MAIL_SENDS") + h.Set(DeadHeaderConsumer, "novox_mesh-mail_send") + h.Set(DeadHeaderSubject, "mesh.stream.mesh-mail.event.mail.send.approved") + h.Set(DeadHeaderRestricted, "true") + h.Set(DeadHeaderWorkQueue, "true") + h.Set(DeadHeaderAgesOut, agesOut.Format(time.RFC3339Nano)) + + d := deadLetterOf(42, "mesh.stream.mesh-mail.event.mail.send.approved", h, []byte("secret body"), false) + if !d.WorkQueue { + t.Fatalf("a work-queue dead letter read as %+v", d) + } + if !d.Restricted { + t.Fatalf("a work-queue dead letter is not read as restricted, so it would be shown with its body: %+v", d) + } + if !d.AgesOut.Equal(agesOut) { + t.Fatalf("ages out at %v, want %v — it should age with its stream", d.AgesOut, agesOut) + } +} + +// 1178/AC3: asking to deliver a work-queue dead letter again is refused, naming ADR 0436 clause 6 — before +// anything is asked of the bus, so it holds whatever max-deliver or age is left. +func TestAWorkQueueDeadLetterIsNeverDeliveredAgain(t *testing.T) { + d := DeadLetter{ID: 7, Stream: "MAIL_SENDS", Consumer: "novox_mesh-mail_send", + Subject: "mesh.stream.mesh-mail.event.mail.send.approved", WorkQueue: true, Restricted: true, + AgesOut: time.Now().Add(time.Hour)} + if _, err := AgainTo(nil, d); err == nil || !strings.Contains(err.Error(), "ADR 0436 clause 6") || + !strings.Contains(err.Error(), "Drop it") { + t.Fatalf("delivering a work-queue dead letter again: %v; want a refusal naming ADR 0436 clause 6", err) + } +}