The bus on NATS: both transports behind seams, and the rollout switch #87
@@ -0,0 +1,134 @@
|
||||
package inventory
|
||||
|
||||
import (
|
||||
"context"
|
||||
"fmt"
|
||||
|
||||
"github.com/novox/mesh-controller/internal/broker"
|
||||
"github.com/novox/mesh-controller/internal/catalogue"
|
||||
)
|
||||
|
||||
// What the bus's user list is derived from, read out of the mesh's records.
|
||||
//
|
||||
// The deriving itself is pure and lives in the broker package; this is the reading, and it is kept
|
||||
// apart for the reason that package keeps its own types: a permission must be a function of what a
|
||||
// module declared, and a query that decided anything would be a second place authority came from.
|
||||
|
||||
// BusRecords is every fact the composer needs about who may reach the bus.
|
||||
//
|
||||
// **A module's authority comes from the manifest, not from the assignment.** The assignment says
|
||||
// *where* it runs; what it may say is in what it declared, so the two are read together and the
|
||||
// manifest is the one that decides.
|
||||
func (i *Inventory) BusRecords(ctx context.Context) (broker.Records, error) {
|
||||
nodes, err := i.Nodes(ctx)
|
||||
if err != nil {
|
||||
return broker.Records{}, fmt.Errorf("cannot read the mesh's machines: %w", err)
|
||||
}
|
||||
declared, err := i.Catalogue(ctx)
|
||||
if err != nil {
|
||||
return broker.Records{}, fmt.Errorf("cannot read the catalogue: %w", err)
|
||||
}
|
||||
|
||||
// Every seat any module declares, by name, so a module's claim can be resolved to the protocol
|
||||
// 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{}
|
||||
for _, m := range declared {
|
||||
for _, s := range m.Seats {
|
||||
seats[s.Name] = s
|
||||
}
|
||||
}
|
||||
|
||||
out := broker.Records{Assigned: map[string][]broker.Declared{}, People: map[string][]string{}}
|
||||
for _, n := range nodes {
|
||||
out.Nodes = append(out.Nodes, n.Name)
|
||||
modules, err := i.Assigned(ctx, n.Name)
|
||||
if err != nil {
|
||||
return broker.Records{}, fmt.Errorf("cannot read what %s runs: %w", n.Name, err)
|
||||
}
|
||||
for _, module := range modules {
|
||||
m, known := declared[module]
|
||||
if !known {
|
||||
// Assigned and not in the catalogue. Said rather than composed with no authority:
|
||||
// a user with an empty permission list is a module that starts, connects, and is
|
||||
// refused by the server on its first publish — an authorisation error that says
|
||||
// nothing about a missing manifest.
|
||||
//
|
||||
// **The catalogue refuses to forget an assigned module, so this is the second line
|
||||
// and not the first.** It earns its place there anyway: relying on another
|
||||
// package's invariant is how a rule ends up enforced by nothing.
|
||||
return broker.Records{}, fmt.Errorf(
|
||||
"%s is assigned to %s and is not in the catalogue, so what it may say cannot "+
|
||||
"be derived", module, n.Name)
|
||||
}
|
||||
out.Assigned[n.Name] = append(out.Assigned[n.Name], declaredFor(m, seats))
|
||||
}
|
||||
}
|
||||
|
||||
enrolling, err := i.NodesWithALiveToken(ctx)
|
||||
if err != nil {
|
||||
return broker.Records{}, err
|
||||
}
|
||||
out.Enrolling = enrolling
|
||||
|
||||
// People are not recorded yet: the account model is built (design 25 §7's first item) and
|
||||
// `operator issue` is not, so there is nobody to derive. Left empty rather than guessed at.
|
||||
return out, nil
|
||||
}
|
||||
|
||||
// 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 {
|
||||
d := broker.Declared{
|
||||
Module: m.Module,
|
||||
Emits: m.Emits,
|
||||
Consumes: m.Consumes,
|
||||
// The tools it answers, which is `tools` and not `serves`: the manifest's `serves` is the
|
||||
// facts a consumer needs to reach a provision, a different meaning under a similar word.
|
||||
Serves: m.Tools,
|
||||
}
|
||||
for _, c := range m.Claims {
|
||||
// A seat the mesh defines for itself declares no protocol, so holding one grants nothing
|
||||
// here — which is right: those seats say who does a job, not who may say what.
|
||||
if s, declaredSomewhere := seats[c.Name]; declaredSomewhere {
|
||||
d.Holds = append(d.Holds, asSeat(s))
|
||||
}
|
||||
}
|
||||
for _, name := range m.Uses {
|
||||
if s, declaredSomewhere := seats[name]; declaredSomewhere {
|
||||
d.Uses = append(d.Uses, asSeat(s))
|
||||
}
|
||||
}
|
||||
return d
|
||||
}
|
||||
|
||||
func asSeat(s catalogue.SeatDeclaration) broker.Seat {
|
||||
return broker.Seat{Name: s.Name, Accepts: s.Accepts, Emits: s.Emits, Serves: s.Serves}
|
||||
}
|
||||
|
||||
// NodesWithALiveToken is every machine holding a token that could still be presented — issued, not
|
||||
// expired, not redeemed.
|
||||
//
|
||||
// **One enrolment user per such token** (design 25 §6): the inbox an answer goes to is scoped to the
|
||||
// token, because an answer carries that machine's credentials sealed to it and a shared inbox is one
|
||||
// machine able to read another's.
|
||||
func (i *Inventory) NodesWithALiveToken(ctx context.Context) ([]string, error) {
|
||||
rows, err := i.store.Pool().Query(ctx,
|
||||
`select distinct n.name
|
||||
from enrolment_token t join node n on n.id = t.node
|
||||
where t.redeemed is null and t.expires > now()
|
||||
order by n.name`)
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("cannot read which machines hold a live token: %w", err)
|
||||
}
|
||||
defer rows.Close()
|
||||
var out []string
|
||||
for rows.Next() {
|
||||
var name string
|
||||
if err := rows.Scan(&name); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
out = append(out, name)
|
||||
}
|
||||
return out, rows.Err()
|
||||
}
|
||||
@@ -0,0 +1,164 @@
|
||||
package inventory
|
||||
|
||||
import (
|
||||
"context"
|
||||
"strings"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
"github.com/novox/mesh-controller/internal/broker"
|
||||
"github.com/novox/mesh-controller/internal/catalogue"
|
||||
)
|
||||
|
||||
// Reading the bus's user list out of the mesh's records, against a real store.
|
||||
//
|
||||
// What each of these is about is a user that would be **missing or wrong in a way nothing reports**:
|
||||
// the server reads whatever file it is given, and a module whose user is absent fails on its first
|
||||
// publish with an authorisation error that says nothing about a missing assignment.
|
||||
|
||||
func aMeshWith(t *testing.T, manifests ...catalogue.Manifest) (*Inventory, context.Context) {
|
||||
t.Helper()
|
||||
inv := ForTest(t)
|
||||
ctx := context.Background()
|
||||
for _, m := range manifests {
|
||||
if err := inv.RegisterModule(ctx, m, Source{Repository: "/r"}); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
}
|
||||
return inv, ctx
|
||||
}
|
||||
|
||||
func theSeatDeclarer() catalogue.Manifest {
|
||||
return catalogue.Manifest{
|
||||
Module: "telegram", Version: "1",
|
||||
Seats: []catalogue.SeatDeclaration{{
|
||||
Name: "telegram-sender", Accepts: []string{"send"}, Emits: []string{"delivered"},
|
||||
}},
|
||||
Claims: []catalogue.Claim{{Name: "telegram-sender", Scope: catalogue.ScopeMesh}},
|
||||
}
|
||||
}
|
||||
|
||||
// A module assigned to a machine becomes a user with the authority its manifest declared — and the
|
||||
// protocol of a seat declared by a *different* module, which is the whole reason a seat exists.
|
||||
func TestAnAssignedModuleBecomesAUserWithWhatItDeclared(t *testing.T) {
|
||||
shop := catalogue.Manifest{
|
||||
Module: "shop", Version: "1",
|
||||
Emits: []string{"order.placed"}, Tools: []string{"price"},
|
||||
Uses: []string{"telegram-sender"},
|
||||
}
|
||||
inv, ctx := aMeshWith(t, theSeatDeclarer(), shop)
|
||||
if _, err := inv.AddNode(ctx, "one"); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if err := inv.Assign(ctx, "one", "shop"); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
|
||||
records, err := inv.BusRecords(ctx)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
on := records.Assigned["one"]
|
||||
if len(on) != 1 || on[0].Module != "shop" {
|
||||
t.Fatalf("the machine's modules read as %+v", on)
|
||||
}
|
||||
if len(on[0].Uses) != 1 || on[0].Uses[0].Accepts[0] != "send" {
|
||||
t.Fatalf("the seat it uses carries no protocol: %+v — so it would be granted nothing on a "+
|
||||
"seat it was assigned to send to", on[0].Uses)
|
||||
}
|
||||
if len(on[0].Serves) != 1 || on[0].Serves[0] != "price" {
|
||||
t.Fatalf("its tools read as %v, and a module that cannot subscribe its own tool subject "+
|
||||
"serves nothing", on[0].Serves)
|
||||
}
|
||||
|
||||
// And it derives into a user the server would accept.
|
||||
users, err := broker.Users(records)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
var found bool
|
||||
for _, u := range users {
|
||||
if u.Username() != "one.shop" {
|
||||
continue
|
||||
}
|
||||
found = true
|
||||
perms, err := broker.PermissionsFor(u)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if !granted(perms.Publish, "mesh.mod.shop.event.order.placed") ||
|
||||
!granted(perms.Publish, "mesh.seat.telegram-sender.accept.send") ||
|
||||
!granted(perms.Subscribe, "mesh.mod.shop.tool.price") {
|
||||
t.Fatalf("one.shop's authority is not what it declared: %+v", perms)
|
||||
}
|
||||
}
|
||||
if !found {
|
||||
t.Fatal("no user was derived for the assigned module")
|
||||
}
|
||||
}
|
||||
|
||||
// A machine holding a live token gets an enrolment user; one whose token is spent or expired does
|
||||
// not. **An enrolment user outliving its token is a right to join that nobody issued.**
|
||||
func TestOnlyAMachineWithALiveTokenHasAnEnrolmentUser(t *testing.T) {
|
||||
inv, ctx := aMeshWith(t)
|
||||
for _, name := range []string{"live", "expired", "none"} {
|
||||
if _, err := inv.AddNode(ctx, name); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
}
|
||||
if _, err := inv.IssueToken(ctx, "live", time.Hour); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
// Briefly, then waited out: a token with no lifetime is refused at issue, which is the right
|
||||
// refusal and leaves this as the way to have an expired one.
|
||||
if _, err := inv.IssueToken(ctx, "expired", 10*time.Millisecond); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
time.Sleep(50 * time.Millisecond)
|
||||
|
||||
records, err := inv.BusRecords(ctx)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if strings.Join(records.Enrolling, ",") != "live" {
|
||||
t.Fatalf("machines with a live token read as %v", records.Enrolling)
|
||||
}
|
||||
}
|
||||
|
||||
// A module assigned and absent from the catalogue is refused rather than composed with no authority.
|
||||
//
|
||||
// **The catalogue refuses to forget an assigned module, so this is the second line and not the
|
||||
// first** — and it earns its place there: relying on another package's invariant is how a rule ends
|
||||
// up enforced by nothing. Checked against the derivation directly, because the situation cannot be
|
||||
// reached through the store.
|
||||
func TestAnAssignmentWithNoManifestDerivesNoAuthority(t *testing.T) {
|
||||
// What BusRecords would have produced had it composed a ghost: a module with nothing declared.
|
||||
users, err := broker.Users(broker.Records{
|
||||
Nodes: []string{"one"},
|
||||
Assigned: map[string][]broker.Declared{"one": {{Module: "ghost"}}},
|
||||
})
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
perms, err := broker.PermissionsFor(users[len(users)-1])
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
// Its inbox and its ack subject, and nothing it could say. That is a module which starts,
|
||||
// connects, and is refused by the server on its first publish — an authorisation error that
|
||||
// says nothing about a missing manifest, which is why BusRecords names it instead.
|
||||
for _, p := range perms.Publish {
|
||||
if strings.HasPrefix(p, "mesh.mod.ghost.event.") {
|
||||
t.Fatalf("a module with no manifest was granted %s", p)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
func granted(all []string, one string) bool {
|
||||
for _, s := range all {
|
||||
if s == one {
|
||||
return true
|
||||
}
|
||||
}
|
||||
return false
|
||||
}
|
||||
Reference in New Issue
Block a user