Grant a seat's traffic by caller and by kind, so an ask's asker and a channel's kind are facts the bus enforces (hq ADR 0259)

This commit is contained in:
jochen
2026-10-09 13:48:56 +02:00
committed by jschoubben
parent 63e85b25e6
commit f5f315cc95
11 changed files with 831 additions and 12 deletions
+32
View File
@@ -53,9 +53,26 @@ func assertBusObjects(ctx context.Context, inv *inventory.Inventory, r broker.Ra
if err != nil {
return nil, err
}
// And the work queues of seats that name their caller or their kind, with each holder's worker
// (novox/hq ADR 0259 §3): an ask queues until the router takes it, a channel's work until that kind
// takes it.
trafficStreams, trafficWorkers, err := seatTrafficObjects(ctx, inv)
if err != nil {
return nil, err
}
for _, s := range trafficStreams {
if err := r.EnsureStream(s); err != nil {
return nil, fmt.Errorf("asserting the work queue %s: %w", s.Name, err)
}
}
// Every one tried, and every failure named: one module's consumer the bus refuses is no reason
// the modules after it in the list hear nothing (novox/hq issue 208, where this runs on each send).
var failed []error
for _, c := range trafficWorkers {
if err := r.EnsureConsumer(c); err != nil {
failed = append(failed, fmt.Errorf("the worker %s on %s: %w", c.Name, c.Stream, err))
}
}
for _, c := range consumers {
if err := r.EnsureConsumer(c.Consumer); err != nil {
failed = append(failed, fmt.Errorf("how %s on %s hears what it consumes: %w", c.Module, c.Node, err))
@@ -130,6 +147,21 @@ func moduleConsumers(ctx context.Context, inv *inventory.Inventory) ([]broker.Mo
return broker.ConsumersOf(users), nil
}
// seatTrafficObjects is the work queues and workers of seats that name their caller or their kind, from
// the records the user list is composed from.
func seatTrafficObjects(ctx context.Context, inv *inventory.Inventory) ([]broker.Stream, []broker.Consumer, error) {
records, err := inv.BusRecords(ctx)
if err != nil {
return nil, nil, err
}
users, err := broker.Users(records)
if err != nil {
return nil, nil, err
}
streams, workers := broker.SeatTrafficObjects(users)
return streams, workers, nil
}
// moduleConsumerCount is how many modules hear what they consume, for the raise's one line.
func moduleConsumerCount(ctx context.Context, inv *inventory.Inventory) (int, error) {
consumers, err := moduleConsumers(ctx, inv)
+3 -1
View File
@@ -126,7 +126,9 @@ func ConsumerFor(p Principal) (Consumer, bool) {
// A module that reacts to anything — a module's events or a role's (novox/hq ADR 0121). Watching
// a role was missing here, so the one module that does it got no consumer at all: it started,
// connected, and its graph stayed empty with nothing anywhere reporting why.
if p.Kind != KindModule || (len(p.Consumes) == 0 && len(p.Watches) == 0) {
// And one that hears its own answers on a seat it uses (novox/hq ADR 0259 §3): an asker's warrants.
hearsItsOwn := len(SeatTrafficOf(p.Module, nil, p.Uses, nil).Subscribe) > 0
if p.Kind != KindModule || (len(p.Consumes) == 0 && len(p.Watches) == 0 && !hearsItsOwn) {
return Consumer{}, false
}
perms, err := PermissionsFor(p)
+10
View File
@@ -50,6 +50,12 @@ type Membership struct {
// and refuses, with the reason, what is not on it — the bus enforces only the union over every
// module on the machine.
State []StateIssued `json:"state,omitempty"`
// SeatTraffic is what this module's code may submit, say, hear, take, ask, answer and read on seats
// that name their caller or their kind (novox/hq ADR 0259 §3). The runtime carrying the module
// publishes, takes and answers for it only what is listed here: the bus enforces only the union
// over every module on the machine, so one module's code reaching another's name or kind through
// the runtime is the runtime's to refuse.
SeatTraffic *SeatTraffic `json:"seat-traffic,omitempty"`
}
// Served is one address a tool is answered on.
@@ -109,6 +115,10 @@ func MembershipFor(node string, d Declared, where Placements) Membership {
}
}
m.State = stateIssuedFor(d, node)
if t := SeatTrafficOf(d.Module, d.Holds, d.Uses, d.Watches); len(t.Publish)+len(t.Subscribe)+
len(t.Answers)+len(t.Workers)+len(t.Records) > 0 {
m.SeatTraffic = &t
}
if len(d.Invokes) > 0 {
m.Reaches = map[string][]string{}
for _, t := range d.Invokes {
+38 -1
View File
@@ -63,6 +63,18 @@ type Seat struct {
Emits []string
Serves []string
Versions []string // protocol versions served beside the current one; empty for v1 only
// Kinded says the seat is a kinded bench (novox/hq ADR 0234 §2, ADR 0259 §3): each holder claims one
// kind, and its verbs' subjects carry it. Kind is the kind this principal's claim names, for a seat it
// holds.
Kinded bool
Kind string
// ByCaller are the accepts and emits whose last token names the calling module (ADR 0259 §3).
ByCaller []string
// Proofs are the seat's proof verbs: core request and reply, never on a stream (ADR 0259 §3).
Proofs []string
// Records are the holder's buckets, by their full name, each user reads under its own name.
Records []string
}
// A Principal is one user of the bus. Its permissions are derived from what it declares and
@@ -530,6 +542,9 @@ func PermissionsFor(p Principal) (Permissions, error) {
// 2b. Events of a role it watches, under the seat's own namespace. Subscribe only: watching a
// role is hearing what it announced, not taking part in it.
for _, w := range p.Watches {
if w.Kinded {
continue // composed by SeatTrafficOf below
}
for _, e := range w.Emits {
sub = append(sub, seatSubject(w, "event", e))
}
@@ -548,6 +563,13 @@ func PermissionsFor(p Principal) (Permissions, error) {
// 3. Seats it holds: full participation.
for _, s := range p.Holds {
if s.isNewTraffic() {
// Composed by SeatTrafficOf below, worker and all; only its tools are served here.
for _, t := range s.Serves {
sub = append(sub, seatToolSubject(s, t, p.Node))
}
continue
}
// Taking work from the role's queue: the worker consumer every holder shares (asked
// about, pulled from, acknowledged), on the seat's own stream (novox/hq ADR 0190). A
// holder pulls — asks the consumer for its next message, answered on its own inbox —
@@ -590,7 +612,7 @@ func PermissionsFor(p Principal) (Permissions, error) {
// seat's inbound subject and watch other modules' traffic, nor publish its outbound
// events and lie about outcomes (design 29 §2).
for _, s := range p.Uses {
for _, a := range s.Accepts {
for _, a := range plainVerbs(s, s.Accepts) {
pub = append(pub, seatSubject(s, "accept", a))
}
for _, t := range s.Serves {
@@ -603,6 +625,12 @@ func PermissionsFor(p Principal) (Permissions, error) {
pub = append(pub, stateGrants(stateAccess{Module: p.Module, Node: p.Node, Keeps: p.State,
PerMachine: p.PerMachine, Reads: p.Reads, KeyedReads: p.KeyedReads})...)
// 6. Its traffic on seats that name their caller or their kind, ask proofs or keep records
// (novox/hq ADR 0259 §3).
tp, ts := SeatTrafficOf(p.Module, p.Holds, p.Uses, p.Watches).grants()
pub = append(pub, tp...)
sub = append(sub, ts...)
case KindNodeTools:
// **One process serves what every module on the machine would have served for itself**
// (novox/hq ADR 0175). Each carried module's whole tool namespace — the same grant that
@@ -680,6 +708,15 @@ func PermissionsFor(p Principal) (Permissions, error) {
pub = append(pub, stateGrants(stateAccess{Module: d.Module, Node: p.Node, Keeps: stateNames(d.State),
PerMachine: perMachineNames(d.State), Reads: d.Reads, KeyedReads: d.KeyedReads})...)
}
// **And the seat traffic of the modules it carries** (novox/hq ADR 0259 §3): a bundle reaches the
// bus only through its runtime, so the runtime is granted the union. That one module's code does
// not publish under another's name or kind through it is the runtime's to keep, from the seat
// traffic each module's membership lists.
for _, d := range p.Carries {
tp, ts := SeatTrafficOf(d.Module, d.Holds, d.Uses, d.Watches).grants()
pub = append(pub, tp...)
sub = append(sub, ts...)
}
sub = unique(sub)
pub = unique(pub)
}
+253
View File
@@ -0,0 +1,253 @@
package broker
import "sort"
// What a module may say and take on the seats it holds, uses and watches, beyond tools (novox/hq ADR
// 0259 §3, to-be 46 §10).
//
// Three rules are added to the ones a seat always had, each a subject whose last token names who may
// publish it, granted to that publisher alone — the way a node seat's event about a machine carries the
// machine (ADR 0219):
//
// - **A verb named by its caller** (`by-caller`). A user of the seat submits that accept, and hears that
// event, under its own module's name and no other: `accept.ask.<module>`, `event.decided.<module>`. The
// holder takes every caller's accept and says the event to any caller. So an ask's asker is a fact the
// server enforces, and a warrant reaches only the asker it is for.
// - **A kinded bench** (ADR 0234 §2). A holder claims one kind and takes its own kind's accepts, says its
// own kind's events and asks its own kind's proofs, and nothing of another kind; a user submits to any
// kind. Each kind has its own worker on the seat's queue, so a holder that is away keeps its work and
// holds up no other kind.
// - **A proof** (`proofs`): core request and reply on `mesh.seat.<seat>.proof.<verb>.<kind>`. No stream's
// subjects cover it, so what travels there — a code typed by the operator — is never persisted. A kinded
// holder asks with its own kind; the modules that watch the seat answer.
//
// And one read: **a holder's records**, a bucket the seat names, read by each user under its own name only
// (`$KV.<bucket>.<module>.>`), so an asker reads the state of its own asks and no other asker's.
// Worker is one durable consumer a holder pulls a seat's work from.
type Worker struct {
Stream string
Consumer string
Filter string
}
// SeatTraffic is one module's seat traffic beyond tools. Publish and Subscribe are subject patterns, in the
// server's wildcards; the runtime that carries the module checks a bundle's request against them, since
// the runtime's own principal holds the union of every module it carries.
type SeatTraffic struct {
// Publish is what it submits (accepts of seats it uses), says (events of seats it holds) and asks
// (proofs of seats it holds a kind of).
Publish []string `json:"publish,omitempty"`
// Subscribe is what it hears: events of seats it uses that are named by caller, events of seats it
// watches, and accepts of seats it holds.
Subscribe []string `json:"subscribe,omitempty"`
// Answers is the proof subjects it answers, as a watcher of a kinded seat.
Answers []string `json:"answers,omitempty"`
// Workers are the work queues it takes from, as a holder.
Workers []Worker `json:"workers,omitempty"`
// Records is the direct-get subjects of the records it reads under its own name.
Records []string `json:"records,omitempty"`
}
// KindedBenches are the seats that may be kinded (ADR 0234 §2): making another is a decision, recorded.
var KindedBenches = map[string]bool{"channel": true, "intake": true}
// WorkerName is the worker a seat's holders pull from: one for the seat, or one per kind on a kinded bench.
func WorkerName(seat, kind string) string {
if kind == "" {
return "SEAT_" + upperSnake(seat) + "_worker"
}
return "SEAT_" + upperSnake(seat) + "_" + upperSnake(kind) + "_worker"
}
func namesVerb(list []string, s string) bool {
for _, x := range list {
if x == s {
return true
}
}
return false
}
// SeatTrafficOf derives one module's seat traffic from the seats it holds, uses and watches. Only seats
// carrying one of the rules above are read: every other seat is composed as it always was.
func SeatTrafficOf(module string, holds, uses, watches []Seat) SeatTraffic {
var t SeatTraffic
for _, s := range holds {
if !s.isNewTraffic() {
continue
}
kind := ""
if s.Kinded {
kind = s.Kind
if kind == "" || !safeSubject.MatchString(kind) {
// A kinded claim without a usable kind is refused at registration; here it is granted
// nothing, which is the same answer at the last place it could be asked.
continue
}
}
if len(s.Accepts) > 0 {
stream := seatStreamName(s.Name)
filter := "mesh.seat." + s.Name + ".accept.>"
if kind != "" {
filter = "mesh.seat." + s.Name + ".accept.*." + kind
}
t.Workers = append(t.Workers, Worker{Stream: stream, Consumer: WorkerName(s.Name, kind), Filter: filter})
}
for _, a := range s.Accepts {
switch {
case kind != "":
t.Subscribe = append(t.Subscribe, seatSubject(s, "accept", a+"."+kind))
case namesVerb(s.ByCaller, a):
t.Subscribe = append(t.Subscribe, seatSubject(s, "accept", a+".*"))
}
}
for _, e := range s.Emits {
switch {
case kind != "":
t.Publish = append(t.Publish, seatSubject(s, "event", e+"."+kind))
case namesVerb(s.ByCaller, e):
t.Publish = append(t.Publish, seatSubject(s, "event", e+".*"))
}
}
if kind != "" {
for _, v := range s.Proofs {
t.Publish = append(t.Publish, seatSubject(s, "proof", v+"."+kind))
}
}
}
for _, s := range uses {
if !s.isNewTraffic() {
continue
}
for _, a := range s.Accepts {
switch {
case namesVerb(s.ByCaller, a):
t.Publish = append(t.Publish, seatSubject(s, "accept", a+"."+module))
case s.Kinded:
t.Publish = append(t.Publish, seatSubject(s, "accept", a+".*"))
}
}
for _, e := range s.Emits {
if namesVerb(s.ByCaller, e) {
t.Subscribe = append(t.Subscribe, seatSubject(s, "event", e+"."+module))
}
}
for _, b := range s.Records {
if !safeSubject.MatchString(b) {
continue
}
t.Records = append(t.Records, "$JS.API.DIRECT.GET.KV_"+b+".$KV."+b+"."+module+".>")
}
}
for _, w := range watches {
if w.Kinded {
for _, e := range w.Emits {
t.Subscribe = append(t.Subscribe, seatSubject(w, "event", e+".*"))
}
for _, v := range w.Proofs {
t.Answers = append(t.Answers, seatSubject(w, "proof", v+".*"))
}
}
}
t.Publish = unique(t.Publish)
t.Subscribe = unique(t.Subscribe)
t.Answers = unique(t.Answers)
t.Records = unique(t.Records)
sort.Slice(t.Workers, func(i, j int) bool { return t.Workers[i].Consumer < t.Workers[j].Consumer })
return t
}
// grants is the bus permissions seat traffic needs: the subjects themselves, and the JetStream API a
// worker is pulled and acknowledged through and a record is read through.
func (t SeatTraffic) grants() (pub, sub []string) {
pub = append(pub, t.Publish...)
sub = append(sub, t.Subscribe...)
sub = append(sub, t.Answers...)
for _, w := range t.Workers {
pub = append(pub,
"$JS.API.CONSUMER.INFO."+w.Stream+"."+w.Consumer,
"$JS.API.CONSUMER.MSG.NEXT."+w.Stream+"."+w.Consumer,
"$JS.ACK."+w.Stream+"."+w.Consumer+".>")
}
pub = append(pub, t.Records...)
return pub, sub
}
// isNewTraffic says whether a seat carries any of the rules above, so a seat that carries none is
// composed exactly as before them.
func (s Seat) isNewTraffic() bool {
return s.Kinded || len(s.ByCaller) > 0 || len(s.Proofs) > 0 || len(s.Records) > 0
}
// plainVerbs is a seat's accepts or emits with those the rules above compose taken out: a verb named by
// its caller and every verb of a kinded bench are composed by SeatTrafficOf and nowhere else.
func plainVerbs(s Seat, verbs []string) []string {
if s.Kinded {
return nil
}
var out []string
for _, v := range verbs {
if !namesVerb(s.ByCaller, v) {
out = append(out, v)
}
}
return out
}
// SeatTrafficObjects is the work queues and workers the seat traffic of every composed user implies
// (novox/hq ADR 0259 §3): a queue for each seat a holder takes work from, and each holder's worker on it —
// one per kind on a kinded bench, filtered to that kind, so the kinds never take each other's work. Only
// seats carrying the rules above; the mesh's own seats' queues are RaiseSeats'.
func SeatTrafficObjects(users []Principal) ([]Stream, []Consumer) {
streams := map[string]Stream{}
consumers := map[string]Consumer{}
add := func(module string, holds []Seat) {
for _, w := range SeatTrafficOf(module, holds, nil, nil).Workers {
seat := ""
for _, s := range holds {
if seatStreamName(s.Name) == w.Stream {
seat = s.Name
}
}
streams[w.Stream] = Stream{
Name: w.Stream,
Subjects: []string{"mesh.seat." + seat + ".accept.>"},
Retention: RetentionWorkQueue,
MaxAge: 7 * 24 * 60 * 60,
Why: "work submitted to the " + seat + " seat; its holders take it, each kind its own, " +
"and it queues while nobody does",
}
consumers[w.Consumer] = Consumer{
Name: w.Consumer,
Stream: w.Stream,
Filters: []string{w.Filter},
AckWaitSeconds: 60,
MaxDeliver: 5,
Why: module + " holds " + seat + "; it pulls one ask at a time and acknowledges once it has " +
"recorded it, so a crash redelivers rather than loses",
}
}
}
for _, p := range users {
switch p.Kind {
case KindModule:
add(p.Module, p.Holds)
case KindNodeTools:
for _, d := range p.Carries {
add(d.Module, d.Holds)
}
}
}
var ss []Stream
for _, s := range streams {
ss = append(ss, s)
}
sort.Slice(ss, func(i, j int) bool { return ss[i].Name < ss[j].Name })
var cs []Consumer
for _, c := range consumers {
cs = append(cs, c)
}
sort.Slice(cs, func(i, j int) bool { return cs[i].Name < cs[j].Name })
return ss, cs
}
+240
View File
@@ -0,0 +1,240 @@
package broker
import (
"strings"
"testing"
)
// The seats of novox/hq ADR 0259 §3, as the messenger declares them.
func operatorChannel() Seat {
return Seat{Name: "operator-channel", Scope: "mesh", Accepts: []string{"ask", "cancel"},
Emits: []string{"decided"}, Serves: []string{"open", "history", "notify"},
ByCaller: []string{"ask", "cancel", "decided"}, Records: []string{"messenger_asks"}}
}
func channelSeat(kind string) Seat {
return Seat{Name: "channel", Scope: "mesh", Accepts: []string{"show", "edit", "send"}, Kinded: true, Kind: kind}
}
func intakeSeat(kind string) Seat {
return Seat{Name: "intake", Scope: "mesh", Emits: []string{"choice", "link"}, Proofs: []string{"code"},
Kinded: true, Kind: kind}
}
func allowed(patterns []string, subject string) bool {
for _, p := range patterns {
if subjectMatches(p, subject) {
return true
}
}
return false
}
func perms(t *testing.T, p Principal) Permissions {
t.Helper()
got, err := PermissionsFor(p)
if err != nil {
t.Fatal(err)
}
return got
}
func TestAnAskerAsksAndHearsUnderItsOwnNameOnly(t *testing.T) {
asker := Principal{Kind: KindModule, Node: "anchor", Module: "mesh-delivery", Uses: []Seat{operatorChannel()}}
got := perms(t, asker)
for _, s := range []string{
"mesh.seat.operator-channel.accept.ask.mesh-delivery",
"mesh.seat.operator-channel.accept.cancel.mesh-delivery",
"$JS.API.DIRECT.GET.KV_messenger_asks.$KV.messenger_asks.mesh-delivery.a1",
} {
if !allowed(got.Publish, s) {
t.Errorf("an asker may not publish %s", s)
}
}
for _, s := range []string{
"mesh.seat.operator-channel.accept.ask.mesh-controller",
"mesh.seat.operator-channel.accept.ask.*",
"mesh.seat.operator-channel.event.decided.mesh-delivery",
"$JS.API.DIRECT.GET.KV_messenger_asks.$KV.messenger_asks.mesh-controller.a1",
"$KV.messenger_asks.mesh-delivery.a1",
} {
if allowed(got.Publish, s) {
t.Errorf("an asker may publish %s, which is not its own to submit", s)
}
}
if !allowed(got.Subscribe, "mesh.seat.operator-channel.event.decided.mesh-delivery") {
t.Error("an asker does not hear its own warrants")
}
if allowed(got.Subscribe, "mesh.seat.operator-channel.event.decided.mesh-controller") {
t.Error("an asker hears another asker's warrants")
}
// And its own consumer carries its warrants, so a restart catches up.
c, ok := ConsumerFor(asker)
if !ok || !allowed(c.Filters, "mesh.seat.operator-channel.event.decided.mesh-delivery") {
t.Errorf("the asker's consumer does not carry its warrants: %v", c.Filters)
}
}
func TestOnlyTheHolderPublishesAWarrant(t *testing.T) {
users, err := Users(Records{
Nodes: []string{"anchor"},
Assigned: map[string][]Declared{"anchor": {
{Module: "messenger", Holds: []Seat{operatorChannel()}, Uses: []Seat{channelSeat("")},
Watches: []Seat{{Name: "intake", Emits: []string{"choice", "link"}, Kinded: true, Proofs: []string{"code"}}}},
{Module: "mesh-delivery", Uses: []Seat{operatorChannel()}},
{Module: "telegram", Holds: []Seat{channelSeat("telegram"), intakeSeat("telegram")}},
}},
})
if err != nil {
t.Fatal(err)
}
for _, u := range users {
got := perms(t, u)
says := allowed(got.Publish, "mesh.seat.operator-channel.event.decided.mesh-delivery")
if says != (u.Module == "messenger") {
t.Errorf("%s %s publish a warrant", u.Username(), map[bool]string{true: "may", false: "may not"}[says])
}
}
}
func TestTheHolderTakesEveryCallersAskThroughItsWorker(t *testing.T) {
got := perms(t, Principal{Kind: KindModule, Node: "anchor", Module: "messenger", Holds: []Seat{operatorChannel()}})
if !allowed(got.Subscribe, "mesh.seat.operator-channel.accept.ask.mesh-delivery") {
t.Error("the router does not take an ask")
}
for _, s := range []string{
"$JS.API.CONSUMER.MSG.NEXT.SEAT_OPERATOR_CHANNEL.SEAT_OPERATOR_CHANNEL_worker",
"$JS.ACK.SEAT_OPERATOR_CHANNEL.SEAT_OPERATOR_CHANNEL_worker.x",
"mesh.seat.operator-channel.event.decided.mesh-controller",
} {
if !allowed(got.Publish, s) {
t.Errorf("the router may not publish %s", s)
}
}
if allowed(got.Publish, "mesh.seat.operator-channel.accept.ask.messenger") {
t.Error("the holder may ask its own seat without using it")
}
}
func TestAKindedHolderReachesItsOwnKindAndNoOther(t *testing.T) {
got := perms(t, Principal{Kind: KindModule, Node: "anchor", Module: "telegram",
Holds: []Seat{channelSeat("telegram"), intakeSeat("telegram")}})
for _, s := range []string{
"mesh.seat.intake.event.choice.telegram", "mesh.seat.intake.event.link.telegram",
"mesh.seat.intake.proof.code.telegram",
"$JS.API.CONSUMER.MSG.NEXT.SEAT_CHANNEL.SEAT_CHANNEL_TELEGRAM_worker",
} {
if !allowed(got.Publish, s) {
t.Errorf("telegram may not publish %s", s)
}
}
for _, s := range []string{
"mesh.seat.intake.event.choice.desktop", "mesh.seat.intake.proof.code.desktop",
"mesh.seat.channel.accept.show.telegram",
"$JS.API.CONSUMER.MSG.NEXT.SEAT_CHANNEL.SEAT_CHANNEL_DESKTOP_worker",
"mesh.seat.operator-channel.event.decided.mesh-delivery",
} {
if allowed(got.Publish, s) {
t.Errorf("telegram may publish %s", s)
}
}
if !allowed(got.Subscribe, "mesh.seat.channel.accept.show.telegram") ||
allowed(got.Subscribe, "mesh.seat.channel.accept.show.desktop") {
t.Error("telegram does not take exactly its own kind's work")
}
if allowed(got.Subscribe, "mesh.seat.intake.proof.code.telegram") {
t.Error("a channel answers its own proofs")
}
}
func TestTheWatcherAnswersProofsAndHearsEveryKind(t *testing.T) {
got := perms(t, Principal{Kind: KindModule, Node: "anchor", Module: "messenger",
Uses: []Seat{channelSeat("")},
Watches: []Seat{{Name: "intake", Emits: []string{"choice"}, Kinded: true, Proofs: []string{"code"}}}})
for _, s := range []string{"mesh.seat.intake.event.choice.telegram", "mesh.seat.intake.proof.code.desktop"} {
if !allowed(got.Subscribe, s) {
t.Errorf("the router does not hear %s", s)
}
}
if !allowed(got.Publish, "mesh.seat.channel.accept.show.telegram") {
t.Error("the router cannot send a channel its work")
}
if allowed(got.Publish, "mesh.seat.intake.event.choice.telegram") || allowed(got.Publish, "mesh.seat.intake.proof.code.telegram") {
t.Error("the router may say a channel's answer or proof")
}
}
func TestNoStreamKeepsAProof(t *testing.T) {
users, _ := Users(Records{Nodes: []string{"anchor"}, Assigned: map[string][]Declared{"anchor": {
{Module: "telegram", Holds: []Seat{channelSeat("telegram"), intakeSeat("telegram")}},
{Module: "messenger", Holds: []Seat{operatorChannel()}},
}}})
streams, _ := SeatTrafficObjects(users)
streams = append(streams, MeshStreams()...)
for _, s := range streams {
for _, subject := range s.Subjects {
if subjectMatches(subject, "mesh.seat.intake.proof.code.telegram") {
t.Errorf("%s keeps a proof (%s)", s.Name, subject)
}
}
}
}
func TestEachKindHasAWorkerOfItsOwn(t *testing.T) {
users, _ := Users(Records{Nodes: []string{"anchor"}, Assigned: map[string][]Declared{"anchor": {
{Module: "telegram", Holds: []Seat{channelSeat("telegram")}},
{Module: "desk-channel", Holds: []Seat{channelSeat("desktop")}},
{Module: "messenger", Holds: []Seat{operatorChannel()}},
}}})
streams, workers := SeatTrafficObjects(users)
names := map[string]string{}
for _, w := range workers {
names[w.Name] = strings.Join(w.Filters, ",")
}
want := map[string]string{
"SEAT_CHANNEL_TELEGRAM_worker": "mesh.seat.channel.accept.*.telegram",
"SEAT_CHANNEL_DESKTOP_worker": "mesh.seat.channel.accept.*.desktop",
"SEAT_OPERATOR_CHANNEL_worker": "mesh.seat.operator-channel.accept.>",
}
for n, f := range want {
if names[n] != f {
t.Errorf("worker %s filters %q, want %q", n, names[n], f)
}
}
if len(streams) != 2 {
t.Errorf("want the queues of channel and operator-channel, got %v", streams)
}
}
func TestTheRuntimeIsGrantedTheUnionAndTheMembershipEachModulesShare(t *testing.T) {
telegram := Declared{Module: "telegram", Holds: []Seat{channelSeat("telegram"), intakeSeat("telegram")}}
desk := Declared{Module: "desk-channel", Holds: []Seat{channelSeat("desktop"), intakeSeat("desktop")}}
got := perms(t, Principal{Kind: KindNodeTools, Node: "anchor", Module: RuntimeModule, Carries: []Declared{telegram, desk}})
for _, s := range []string{"mesh.seat.intake.event.choice.telegram", "mesh.seat.intake.event.choice.desktop"} {
if !allowed(got.Publish, s) {
t.Errorf("the runtime may not publish %s for a module it carries", s)
}
}
m := MembershipFor("anchor", telegram, Placements{})
if m.SeatTraffic == nil || !allowed(m.SeatTraffic.Publish, "mesh.seat.intake.event.choice.telegram") ||
allowed(m.SeatTraffic.Publish, "mesh.seat.intake.event.choice.desktop") {
t.Errorf("telegram's membership does not list exactly its own kind: %+v", m.SeatTraffic)
}
if plain := MembershipFor("anchor", Declared{Module: "plain"}, Placements{}); plain.SeatTraffic != nil {
t.Error("a module with no such seat is given seat traffic")
}
}
func TestASeatWithoutTheNewRulesIsComposedAsBefore(t *testing.T) {
old := Seat{Name: "node-build-agent", Scope: "node", Accepts: []string{"build"}, Emits: []string{"built"}}
got := perms(t, Principal{Kind: KindModule, Node: "anchor", Module: "builder", Holds: []Seat{old}})
for _, s := range []string{"mesh.seat.node-build-agent.event.built",
"$JS.API.CONSUMER.MSG.NEXT.SEAT_NODE_BUILD_AGENT.SEAT_NODE_BUILD_AGENT_worker"} {
if !allowed(got.Publish, s) {
t.Errorf("an old seat's holder lost %s", s)
}
}
if !allowed(got.Subscribe, "mesh.seat.node-build-agent.accept.build") {
t.Error("an old seat's holder lost its accept")
}
}
+95
View File
@@ -0,0 +1,95 @@
package catalogue
import (
"strings"
"testing"
)
// The router of novox/hq ADR 0259: it declares the operator's seat and the two kinded benches.
func router() Manifest {
return Manifest{Module: "messenger", Tools: []string{"open", "history", "notify"},
State: []StateDeclaration{{Name: "asks"}},
DefinesSeats: []SeatDeclaration{
{Name: "operator-channel", Scope: ScopeMesh, Accepts: []string{"ask", "cancel"}, Emits: []string{"decided"},
ByCaller: []string{"ask", "cancel", "decided"}, Records: []string{"asks"},
Serves: []Verb{{Name: "open"}, {Name: "history"}, {Name: "notify"}}},
{Name: "channel", Scope: ScopeMesh, Kinded: true, Accepts: []string{"show", "edit", "send"}},
{Name: "intake", Scope: ScopeMesh, Kinded: true, Emits: []string{"choice", "link"}, Proofs: []string{"code"}},
},
Claims: []Claim{{Name: "operator-channel", Scope: ScopeMesh}},
Uses: []string{"channel"}}
}
func aChannel(module, kind string) Manifest {
return Manifest{Module: module, Claims: []Claim{
{Name: "channel", Scope: ScopeMesh, Kind: kind}, {Name: "intake", Scope: ScopeMesh, Kind: kind}}}
}
func TestTwoChannelsOfDifferentKindsHoldTheBenches(t *testing.T) {
shelf := Shelf{"messenger": router(), "telegram": aChannel("telegram", "telegram"),
"desk-channel": aChannel("desk-channel", "desktop")}
if got := problemsFor(t, shelf); got != "" {
t.Fatalf("two kinds were refused: %s", got)
}
for _, m := range shelf {
if got := declaredSeatProblems(m); len(got) > 0 {
t.Fatalf("%s: %v", m.Module, got)
}
}
}
func TestASecondClaimOfOneKindIsRefused(t *testing.T) {
got := problemsFor(t, Shelf{"messenger": router(), "telegram": aChannel("telegram", "telegram"),
"telegram-two": aChannel("telegram-two", "telegram")})
if !strings.Contains(got, `of kind "telegram", which telegram already claims`) {
t.Fatalf("a second holder of one kind stood: %s", got)
}
}
func TestAKindedBenchNeedsAKindAndNoOtherSeatTakesOne(t *testing.T) {
got := problemsFor(t, Shelf{"messenger": router(), "nameless": aChannel("nameless", "")})
if !strings.Contains(got, "claims the kinded bench channel and names no kind") {
t.Fatalf("a claim without a kind stood: %s", got)
}
odd := Manifest{Module: "odd", Claims: []Claim{{Name: "operator-channel", Scope: ScopeMesh, Kind: "telegram"}}}
got = problemsFor(t, Shelf{"messenger": router(), "odd": odd})
if !strings.Contains(got, "only a kinded bench takes a kind") {
t.Fatalf("a kind on a seat that is not kinded stood: %s", got)
}
dotted := problemsFor(t, Shelf{"messenger": router(), "dotted": aChannel("dotted", "a.b")})
if !strings.Contains(dotted, "not a usable name") {
t.Fatalf("a kind that would widen a subject stood: %s", dotted)
}
}
func TestOnlyChannelAndIntakeAreKinded(t *testing.T) {
m := Manifest{Module: "x", DefinesSeats: []SeatDeclaration{{Name: "pager", Kinded: true, Accepts: []string{"page"}}}}
if got := strings.Join(declaredSeatProblems(m), "; "); !strings.Contains(got, "only channel and intake are kinded") {
t.Fatalf("another kinded bench was declared: %s", got)
}
}
func TestTheNewRulesAreHeldToWhatTheSeatSays(t *testing.T) {
m := Manifest{Module: "x", DefinesSeats: []SeatDeclaration{{Name: "thing", Accepts: []string{"do"},
ByCaller: []string{"undo"}, Proofs: []string{"code"}, Records: []string{"nothing"}}}}
got := strings.Join(declaredSeatProblems(m), "; ")
for _, want := range []string{"names thing.undo by its caller, which the seat neither accepts nor emits",
"declares proofs on thing, which is not kinded", `read its records "nothing", which it keeps no state of`} {
if !strings.Contains(got, want) {
t.Errorf("not refused: %q in %s", want, got)
}
}
}
// Two kinds on one machine are two holders, and one kind on two machines is a second claimant.
func TestEachKindIsItsOwnHolderWhenResolved(t *testing.T) {
modules := []Manifest{aChannel("telegram", "telegram"), aChannel("desk-channel", "desktop")}
held, problems := checkClaims(modules, Node{Name: "anchor"}, nil, nil)
if len(problems) > 0 || len(held) != 4 {
t.Fatalf("two kinds on one machine: held %v, problems %v", held, problems)
}
_, problems = checkClaims([]Manifest{aChannel("telegram", "telegram")}, Node{Name: "home"}, held, nil)
if len(problems) == 0 {
t.Fatal("one kind held on two machines was not refused")
}
}
+3
View File
@@ -64,6 +64,9 @@ type Claim struct {
// text/template over one piece — its fields and `module` — in the tool's own grammar (novox/hq ADR
// 0255). The data is the mesh's, the format the holder's, as a module's facts template is.
Renders map[string]string `json:"renders,omitempty"`
// Kind is the kind this module holds a kinded bench as (novox/hq ADR 0234 §2, ADR 0259): `telegram`,
// `desktop`. Refused on any other seat, and a second claim of one kind is refused.
Kind string `json:"kind,omitempty"`
}
// ServesFor is what this claim offers a seat's protocol: the verbs it names, else the module's
+13 -3
View File
@@ -120,6 +120,16 @@ type Held struct {
Node string
Module string
Site string
// Kind is the kind a kinded bench is held as (novox/hq ADR 0234 §2): each kind is its own holder.
Kind string
}
// heldKey is what one holder holds: the seat, and its kind on a kinded bench.
func heldKey(claim, kind string) string {
if kind == "" {
return canonicalSeat(claim)
}
return canonicalSeat(claim) + "/" + kind
}
// Resolution is what a node should run, and why.
@@ -874,7 +884,7 @@ func checkClaims(modules []Manifest, node Node, elsewhere []Held, holdings []Hel
// **One seat under either of its names** (novox/hq ADR 0122): a manifest registered before
// a rename claims the former name, and one written after it the current — two claimants of
// one seat, compared by the seat they resolve to and not by how each spelled it.
seat := canonicalSeat(c.Name)
seat := heldKey(c.Name, c.Kind)
if other, taken := byScope[scope][seat]; taken {
problems = append(problems, fmt.Sprintf(
"%s and %s both claim %q, and only one thing may hold it per %s",
@@ -883,14 +893,14 @@ func checkClaims(modules []Manifest, node Node, elsewhere []Held, holdings []Hel
}
byScope[scope][seat] = m.Module
held = append(held, Held{Claim: c.Name, Scope: scope, Node: node.Name,
Module: m.Module, Site: node.Site})
Module: m.Module, Site: node.Site, Kind: c.Kind})
}
}
// And against the rest of the mesh, for the scopes that reach past this machine.
for _, h := range held {
for _, e := range elsewhere {
if e.Node == node.Name || canonicalSeat(e.Claim) != canonicalSeat(h.Claim) || e.Scope != h.Scope {
if e.Node == node.Name || heldKey(e.Claim, e.Kind) != heldKey(h.Claim, h.Kind) || e.Scope != h.Scope {
continue
}
switch h.Scope {
+115
View File
@@ -51,6 +51,35 @@ type SeatDeclaration struct {
// owns its own, which is why a seat is also the answer for a module that needs retention
// its events cannot have.
RetainSeconds int `json:"retain-seconds,omitempty"`
// Kinded makes the seat a kinded bench (novox/hq ADR 0234 §2, ADR 0259 §3): its holders are different
// modules, each claiming one kind, and each verb's subject carries the kind. Only the benches in
// KindedBenches may be kinded; making another is a decision, recorded.
Kinded bool `json:"kinded,omitempty"`
// ByCaller are accepts and emits whose subject's last token names the calling module (ADR 0259 §3): a
// user submits such an accept, and hears such an event, under its own name and no other.
ByCaller []string `json:"by-caller,omitempty"`
// Proofs are verbs carried as core request and reply, never on a stream: what travels on them (a code
// the operator typed) is never kept (ADR 0259 §3). On a kinded bench a holder asks with its own kind and
// the modules watching the seat answer.
Proofs []string `json:"proofs,omitempty"`
// Records are state buckets of the declaring module that each user reads under its own name — the
// keys `<user>.…` and no other (ADR 0259 §3).
Records []string `json:"records,omitempty"`
}
// KindedBenches are the seats that may be kinded (novox/hq ADR 0234 §2): `channel` sends to the operator,
// `intake` takes what the operator answers. Another is a decision, recorded, as ADR 0223 asks of a bench.
var KindedBenches = map[string]bool{"channel": true, "intake": true}
// NamedByCaller says whether one of the seat's verbs is named by its caller.
func (s SeatDeclaration) NamedByCaller(verb string) bool {
for _, v := range s.ByCaller {
if v == verb {
return true
}
}
return false
}
// At is this declaration's scope, with the default applied. Mesh by default, because a seat
@@ -132,6 +161,7 @@ func declaredSeatProblems(m Manifest) []string {
"%s declares %s.%s, which is not a usable verb", m.Module, s.Name, v))
}
}
problems = append(problems, trafficProblems(m, s)...)
}
for _, u := range m.Uses {
@@ -142,6 +172,56 @@ func declaredSeatProblems(m Manifest) []string {
return problems
}
// trafficProblems is what one declaration of the rules of ADR 0259 §3 can be judged on alone.
func trafficProblems(m Manifest, s SeatDeclaration) []string {
var problems []string
if s.Kinded && !KindedBenches[s.Name] {
problems = append(problems, fmt.Sprintf(
"%s declares %s as a kinded bench; only channel and intake are kinded, and another is a "+
"decision, recorded (novox/hq ADR 0234)", m.Module, s.Name))
}
if s.Kinded && len(s.ByCaller) > 0 {
problems = append(problems, fmt.Sprintf(
"%s declares %s kinded and names verbs by their caller; a kinded bench's subjects carry the kind",
m.Module, s.Name))
}
for _, v := range s.ByCaller {
inAccepts, inEmits := false, false
for _, a := range s.Accepts {
inAccepts = inAccepts || a == v
}
for _, e := range s.Emits {
inEmits = inEmits || e == v
}
if !inAccepts && !inEmits {
problems = append(problems, fmt.Sprintf(
"%s names %s.%s by its caller, which the seat neither accepts nor emits", m.Module, s.Name, v))
}
}
for _, v := range s.Proofs {
if !name.MatchString(v) || strings.Contains(v, ".") {
problems = append(problems, fmt.Sprintf("%s declares the proof %s.%s, which is not a usable verb",
m.Module, s.Name, v))
}
}
if len(s.Proofs) > 0 && !s.Kinded {
problems = append(problems, fmt.Sprintf(
"%s declares proofs on %s, which is not kinded; a proof is asked by a holder of a kind",
m.Module, s.Name))
}
for _, r := range s.Records {
kept := false
for _, st := range m.State {
kept = kept || st.Name == r
}
if !kept {
problems = append(problems, fmt.Sprintf(
"%s says %s's users read its records %q, which it keeps no state of", m.Module, s.Name, r))
}
}
return problems
}
// A Shelf is every manifest the mesh has registered, by module name.
type Shelf map[string]Manifest
@@ -182,8 +262,19 @@ func CatalogueProblems(shelf Shelf) []string {
return ok
}
// Who claims each kind of a kinded bench, so a second claim of one kind is refused (ADR 0234 §2).
kindsTaken := map[string]string{}
for _, module := range shelfOrder(shelf) {
m := shelf[module]
for _, c := range m.Claims {
if c.Kind != "" {
if _, isModuleSeat := declared[c.Name]; !isModuleSeat {
problems = append(problems, fmt.Sprintf(
"%s claims %s of kind %q, and only a kinded bench takes a kind", module, c.Name, c.Kind))
}
}
}
// A `uses` naming nothing is where ADR 0110's guarantee lands under a derived set: the
// same refusal, at the same moment, from a set nobody maintains by hand.
@@ -214,6 +305,7 @@ func CatalogueProblems(shelf Shelf) []string {
}
continue
}
problems = append(problems, kindProblems(module, c, s, kindsTaken)...)
if c.At() != s.At() {
problems = append(problems, fmt.Sprintf(
"%s claims %s at scope %q, and %s declares it at %s",
@@ -239,6 +331,29 @@ func CatalogueProblems(shelf Shelf) []string {
return problems
}
// kindProblems is a claim judged against a declared seat's kind: a kinded bench takes one claim per kind,
// a usable name; any other seat takes none.
func kindProblems(module string, c Claim, s SeatDeclaration, taken map[string]string) []string {
switch {
case !s.Kinded && c.Kind != "":
return []string{fmt.Sprintf("%s claims %s of kind %q, and only a kinded bench takes a kind",
module, c.Name, c.Kind)}
case !s.Kinded:
return nil
case c.Kind == "":
return []string{fmt.Sprintf("%s claims the kinded bench %s and names no kind", module, c.Name)}
case !name.MatchString(c.Kind) || strings.Contains(c.Kind, "."):
return []string{fmt.Sprintf("%s claims %s of kind %q, which is not a usable name", module, c.Name, c.Kind)}
}
key := c.Name + "/" + c.Kind
if first, ok := taken[key]; ok && first != module {
return []string{fmt.Sprintf("%s claims %s of kind %q, which %s already claims; a kind has one holder",
module, c.Name, c.Kind, first)}
}
taken[key] = module
return nil
}
// unserved is what a seat's protocol promises and the claimant does not answer. Only the tools
// are checked: `accepts` and `emits` are wired by the runtime from the declaration, while a tool
// is code the module either has or has not written — under the claim's serves, or among its own.
+29 -7
View File
@@ -34,9 +34,12 @@ func (i *Inventory) BusRecords(ctx context.Context) (broker.Records, error) {
// that seat promises. **Across the whole catalogue, not one manifest**: a seat is declared by
// one module and held by another, which is the whole reason a seat exists (ADR 0118).
seats := map[string]catalogue.SeatDeclaration{}
// And who declared each, so a seat's records are named as the declaring module's buckets (ADR 0259).
declarers := map[string]string{}
for _, m := range declared {
for _, s := range m.DefinesSeats {
seats[s.Name] = s
declarers[s.Name] = m.Module
}
}
// And the mesh's own, which carry protocol too (novox/hq ADR 0121). Added after the modules'
@@ -78,7 +81,7 @@ func (i *Inventory) BusRecords(ctx context.Context) (broker.Records, error) {
"%s is assigned to %s and is not in the catalogue, so what it may say cannot "+
"be derived", module, n.Name)
}
d := declaredFor(m, seats)
d := declaredFor(m, seats, declarers)
// And the state offered to it as a seat's holder by the modules beside it (novox/hq ADR 0255).
// For this machine's key alone (novox/hq ADR 0260): a bar shows its own machine's draw.
for _, sr := range catalogue.ReadsGranted(m, onMachine(declared, modules), n.Name) {
@@ -121,7 +124,7 @@ func onMachine(declared map[string]catalogue.Manifest, modules []string) []catal
// declaredFor is one module's manifest as the composer needs it: what it says about itself, and the
// protocol of every seat it holds or uses.
func declaredFor(m catalogue.Manifest, seats map[string]catalogue.SeatDeclaration) broker.Declared {
func declaredFor(m catalogue.Manifest, seats map[string]catalogue.SeatDeclaration, declarers map[string]string) broker.Declared {
// A consumed name is a module's event unless it names a seat, and only somebody holding the seat
// set can tell (novox/hq ADR 0121). Split here, because the composer cannot look at a name and
// know — and a role's event read as a module's is a subscription to a namespace nobody owns.
@@ -132,6 +135,16 @@ func declaredFor(m catalogue.Manifest, seats map[string]catalogue.SeatDeclaratio
if named {
// A seat's event when the seat says it; else the event of the module of that name — a seat and
// the module holding it may share a name (mesh-delivery, novox/hq ADR 0239).
if s, isASeat := seats[emitter]; isASeat && s.Kinded {
// A kinded bench's event is named `<event>` or `<event>.*` and heard from every kind; its
// proofs are answered by whoever watches it (ADR 0259 §3).
ev := strings.TrimSuffix(event, ".*")
if catalogue.SeatSays(s.Emits, ev) {
watches = append(watches, broker.Seat{Name: s.Name, Scope: s.Scope, Emits: []string{ev},
Kinded: true, Proofs: s.Proofs})
continue
}
}
if s, isASeat := seats[emitter]; isASeat && catalogue.SeatSays(s.Emits, event) {
watches = append(watches, broker.Seat{Name: s.Name, Emits: []string{event}})
continue
@@ -168,12 +181,14 @@ func declaredFor(m catalogue.Manifest, seats map[string]catalogue.SeatDeclaratio
// Every seat with a protocol, the mesh's own included. One that says only who does a job is
// not here and grants nothing, which is most of them.
if s, hasAProtocol := seats[c.Name]; hasAProtocol {
d.Holds = append(d.Holds, asSeat(s))
held := asSeat(s, declarers[s.Name])
held.Kind = c.Kind
d.Holds = append(d.Holds, held)
}
}
for _, name := range m.Uses {
if s, declaredSomewhere := seats[name]; declaredSomewhere {
d.Uses = append(d.Uses, asSeat(s))
d.Uses = append(d.Uses, asSeat(s, declarers[s.Name]))
}
}
return d
@@ -204,9 +219,16 @@ func (i *Inventory) DeclaredBuckets(ctx context.Context) ([]broker.Bucket, error
return out, nil
}
func asSeat(s catalogue.SeatDeclaration) broker.Seat {
return broker.Seat{Name: s.Name, Scope: s.Scope, Accepts: s.Accepts, Emits: s.Emits,
Serves: catalogue.VerbNames(s.Serves)}
func asSeat(s catalogue.SeatDeclaration, declarer string) broker.Seat {
seat := broker.Seat{Name: s.Name, Scope: s.Scope, Accepts: s.Accepts, Emits: s.Emits,
Serves: catalogue.VerbNames(s.Serves), Kinded: s.Kinded, ByCaller: s.ByCaller, Proofs: s.Proofs}
// A seat's records are its declaring module's buckets, named as the bus holds them (ADR 0259 §3).
for _, r := range s.Records {
if declarer != "" {
seat.Records = append(seat.Records, broker.BucketName(declarer, r))
}
}
return seat
}
// MeshSeats are the mesh's own seats that carry a protocol, as the bus needs them: what to make a work