Compare commits
37
Commits
@@ -67,6 +67,9 @@ type deskGive struct {
|
||||
// nil keeps no record (a test that does not look). end closes it with how it ended.
|
||||
open func(node, module, name, desk string) (int64, error)
|
||||
end func(id int64, outcome string) error
|
||||
// vault, when set, is the path of a module whose secrets are filled the new way (vault_secrets.go): the vault
|
||||
// records the ask, holds its bounds and opens what is typed; open, end and accept are not used.
|
||||
vault vaultAsker
|
||||
}
|
||||
|
||||
// askedByName is the caller as the prompt names it: the first clause of what the bus said, in the characters
|
||||
@@ -133,6 +136,9 @@ func (d deskGive) give(node, module, name, desk string) (string, error) {
|
||||
"bus, where an agent may answer first (novox/hq ADR 0259 §10)", module, name, node, module, name)
|
||||
}
|
||||
}
|
||||
if d.vault != nil {
|
||||
return d.giveToVault(node, module, name, desk)
|
||||
}
|
||||
// The bounds, read and kept before anybody is asked to type (novox/hq ADR 0277): one open prompt per secret,
|
||||
// few an hour. How the ask ends is recorded whatever happens below.
|
||||
outcome := "failed"
|
||||
@@ -275,6 +281,11 @@ func askAtDesk(ctx context.Context, node, module, name, desk string) error {
|
||||
},
|
||||
announce: func(node, module, name, how string) error { return announceSecretGiven(ctx, node, module, name, how) },
|
||||
}
|
||||
// A module whose secrets are filled the new way is asked for by the vault (ADR 0349 clause 9); every other
|
||||
// keeps today's path (clause 17).
|
||||
if secretsOfTheVault(module) {
|
||||
d.vault = theVault(ctx)
|
||||
}
|
||||
words, err := d.give(node, module, name, desk)
|
||||
if err != nil {
|
||||
return err
|
||||
|
||||
@@ -53,6 +53,10 @@ type issuedTaker struct {
|
||||
keep func(provision, provider, value string) (inventory.IssuedKept, error)
|
||||
record func(act link.HandAct) error
|
||||
send func(machines []string) error
|
||||
// ofTheVault says whether a provider module's secrets are filled the new way; for one that is, the value goes
|
||||
// into the vault (vault, vault_secrets.go) and nothing is kept, sealed or sent here. Nil is none.
|
||||
ofTheVault func(module string) bool
|
||||
vault vaultAsker
|
||||
}
|
||||
|
||||
// take asks every provider of the provision for its software's value and keeps each; it answers what it did, in
|
||||
@@ -90,6 +94,9 @@ func (t issuedTaker) take(provision string) (string, error) {
|
||||
|
||||
// takeFrom is one provider: asked, opened, kept, recorded and sent when it changed.
|
||||
func (t issuedTaker) takeFrom(provision string, p issuedProvider) (string, error) {
|
||||
if t.ofTheVault != nil && t.ofTheVault(p.Module) {
|
||||
return t.takeIntoVault(provision, p)
|
||||
}
|
||||
public, private, err := secrets.Keypair()
|
||||
if err != nil {
|
||||
return "", fmt.Errorf("no key could be made to take the value: %w", err)
|
||||
@@ -262,7 +269,9 @@ func secretIssued(ctx context.Context, args []string) error {
|
||||
return err
|
||||
})
|
||||
},
|
||||
send: func(machines []string) error { return sendTo(ctx, open, machines) },
|
||||
send: func(machines []string) error { return sendTo(ctx, open, machines) },
|
||||
ofTheVault: secretsOfTheVault,
|
||||
vault: theVault(ctx),
|
||||
}
|
||||
words, err := t.take(provision)
|
||||
if words != "" {
|
||||
|
||||
@@ -264,9 +264,6 @@ func serve(ctx context.Context) (err error) {
|
||||
// The facts snapshot a merge check is fed (novox/hq to-be 45 §9): kept current while this
|
||||
// controller holds the lease, read by the build seat from the artifact store.
|
||||
go exportingFacts(ctx, open, bus.Conn.ConnectedServerVersion)
|
||||
// The record mirrors (novox/hq ADR 0289, design 57 F3): the assignments and each registered module's
|
||||
// declared config on the bus, written from the store only when a record changed, by the lease holder.
|
||||
go mirroringRecords(ctx, inv, bus.Conn, bus)
|
||||
// Every call carries the lease's epoch, and its record is written only under the lease (novox/hq
|
||||
// to-be 45 §6).
|
||||
link.Calls.UnderLease(func() (uint64, error) { return theLease.epoch(ctx) })
|
||||
|
||||
@@ -1,105 +0,0 @@
|
||||
package main
|
||||
|
||||
import (
|
||||
"context"
|
||||
"fmt"
|
||||
"time"
|
||||
|
||||
"github.com/nats-io/nats.go"
|
||||
"github.com/nats-io/nats.go/jetstream"
|
||||
|
||||
"github.com/novox/mesh-controller/internal/inventory"
|
||||
"github.com/novox/mesh-controller/internal/recordmirror"
|
||||
)
|
||||
|
||||
// The controller's record mirrors on the bus (novox/hq ADR 0289, design 54, design 57 F3): the assignments,
|
||||
// and each registered module's declared config, which mesh-config resolves a module's config values from.
|
||||
//
|
||||
// **Built from the store, on a tick of their own, by the lease holder alone.** A record changes through a
|
||||
// verb that runs as a command of its own, a reconcile, a merge's registration or a pending assignment made,
|
||||
// so no one call site sees every change; what every change has in common is that it is committed to the
|
||||
// store. So the serving controller reads the store every mirrorEvery and writes only what differs from what
|
||||
// the bucket holds (ADR 0289 §1): a quiet mesh writes nothing, and a change rolled back is never read. A
|
||||
// mirror is therefore at most mirrorEvery behind its store.
|
||||
|
||||
// mirrorEvery is how often the mirrors are compared with the store: two small reads of the store.
|
||||
const mirrorEvery = 10 * time.Second
|
||||
|
||||
// mirroringRecords keeps the mirrors current for as long as this controller serves: ctx ends when it stops.
|
||||
func mirroringRecords(ctx context.Context, inv *inventory.Inventory, conn *nats.Conn, announce recordmirror.Announcer) {
|
||||
api, err := jetstream.New(conn)
|
||||
if err != nil {
|
||||
fmt.Printf("the record mirrors are not kept: %v\n", err)
|
||||
return
|
||||
}
|
||||
holds := func() bool {
|
||||
epoch, err := theLease.epoch(ctx)
|
||||
return err == nil && epoch > 0
|
||||
}
|
||||
epoch := func() uint64 {
|
||||
e, _ := theLease.epoch(ctx)
|
||||
return e
|
||||
}
|
||||
sources := map[string]func(context.Context) (map[string][]byte, error){
|
||||
recordmirror.Assignments.Name: inv.AssignmentsMirror,
|
||||
recordmirror.DeclaredConfig.Name: inv.DeclaredConfigMirror,
|
||||
}
|
||||
writers := map[string]*recordmirror.Writer{}
|
||||
failing := map[string]string{}
|
||||
tick := time.NewTicker(mirrorEvery)
|
||||
defer tick.Stop()
|
||||
for {
|
||||
if holds() {
|
||||
for _, m := range recordmirror.Mirrors() {
|
||||
why := mirrorOnce(ctx, api, m, writers, sources[m.Name], announce, epoch, holds)
|
||||
if why != failing[m.Name] {
|
||||
if why != "" {
|
||||
fmt.Printf("the %s mirror cannot be kept: %s\n", m.Name, why)
|
||||
} else {
|
||||
fmt.Printf("the %s mirror is kept again\n", m.Name)
|
||||
}
|
||||
failing[m.Name] = why
|
||||
}
|
||||
}
|
||||
}
|
||||
select {
|
||||
case <-ctx.Done():
|
||||
return
|
||||
case <-tick.C:
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// mirrorOnce brings one mirror to what the store holds, and answers why it could not; empty when it could.
|
||||
func mirrorOnce(ctx context.Context, api jetstream.JetStream, m recordmirror.Mirror, writers map[string]*recordmirror.Writer,
|
||||
source func(context.Context) (map[string][]byte, error), announce recordmirror.Announcer, epoch func() uint64,
|
||||
holds func() bool) string {
|
||||
reading, cancel := context.WithTimeout(ctx, 30*time.Second)
|
||||
defer cancel()
|
||||
w := writers[m.Name]
|
||||
if w == nil {
|
||||
kv, err := api.KeyValue(reading, m.Bucket)
|
||||
if err != nil {
|
||||
return fmt.Sprintf("the bucket %s cannot be read: %v", m.Bucket, err)
|
||||
}
|
||||
w = &recordmirror.Writer{Mirror: m, KV: kv, Announce: announce, Epoch: epoch, Holds: holds}
|
||||
writers[m.Name] = w
|
||||
}
|
||||
want, err := source(reading)
|
||||
if err != nil {
|
||||
return fmt.Sprintf("the store cannot be read: %v", err)
|
||||
}
|
||||
lost := w.LostEvents
|
||||
r, err := w.Sync(reading, want)
|
||||
if r.Written+r.Removed > 0 {
|
||||
fmt.Printf("the %s mirror moved: %d written, %d removed\n", m.Name, r.Written, r.Removed)
|
||||
}
|
||||
if w.LostEvents > lost {
|
||||
fmt.Printf("the %s mirror lost %d event(s): the bucket holds the change, its notice was refused twice\n",
|
||||
m.Name, w.LostEvents-lost)
|
||||
}
|
||||
if err != nil {
|
||||
return err.Error()
|
||||
}
|
||||
return ""
|
||||
}
|
||||
@@ -108,7 +108,12 @@ func secretCommand(ctx context.Context, args []string) error {
|
||||
if value == "" {
|
||||
return errors.New("there is nothing to seal")
|
||||
}
|
||||
return acceptForProvision(ctx, node, module, name, value)
|
||||
// The provider's module decides the path: filled the new way, the vault keeps the provision's credential.
|
||||
return bySwitch(module, secretsOfTheVault,
|
||||
func() error { return acceptForProvision(ctx, node, module, name, value) },
|
||||
func() error {
|
||||
return said(acceptIntoVault(theVault(ctx), provisionArgs(node, name), value, actingAs()))
|
||||
})
|
||||
}
|
||||
if *desk != "" {
|
||||
if *from != "" || *provider != "" {
|
||||
@@ -125,23 +130,50 @@ func secretCommand(ctx context.Context, args []string) error {
|
||||
if value == "" {
|
||||
return errors.New("there is nothing to seal")
|
||||
}
|
||||
return bySwitch(module, secretsOfTheVault,
|
||||
func() error { return acceptKept(ctx, node, module, name, *provider, *local, value) },
|
||||
func() error {
|
||||
// Sealed here to the vault's key for this one call; only the sealed box leaves this process.
|
||||
target := ownSecretArgs(node, module, name)
|
||||
if *provider != "" {
|
||||
if *local != "" {
|
||||
return errors.New("--local names a credential inside the module; the vault keeps one credential " +
|
||||
"per provider and provision, so it takes no --local: nothing was sealed")
|
||||
}
|
||||
target = provisionArgs(*provider, name)
|
||||
}
|
||||
return said(acceptIntoVault(theVault(ctx), target, value, actingAs()))
|
||||
})
|
||||
}
|
||||
|
||||
// said prints what was done, or answers why not.
|
||||
func said(words string, err error) error {
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
fmt.Println(words)
|
||||
return nil
|
||||
}
|
||||
|
||||
// acceptKept is today's `secret accept`: the value sealed into the controller's store, for a module whose secrets
|
||||
// are filled the old way.
|
||||
func acceptKept(ctx context.Context, node, module, name, provider, local, value string) error {
|
||||
open, err := openStores(ctx)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
defer open.Close()
|
||||
|
||||
if *provider != "" {
|
||||
if provider != "" {
|
||||
// Into the pair, not into the module's own secrets: what the provider is asked to create
|
||||
// and what the consumer reads are the same value, and neither end can be told a different
|
||||
// one later without the other (novox/hq 04-ISSUES/070).
|
||||
if err := open.inventory.AcceptSecretForPair(ctx, name, node, module, *provider, *local, value); err != nil {
|
||||
if err := open.inventory.AcceptSecretForPair(ctx, name, node, module, provider, local, value); err != nil {
|
||||
return err
|
||||
}
|
||||
fmt.Printf("%s on %s now holds %q from %s%s, sealed to both machines.\n", module, node, name, *provider, asLocal(*local))
|
||||
fmt.Printf("%s on %s now holds %q from %s%s, sealed to both machines.\n", module, node, name, provider, asLocal(local))
|
||||
fmt.Printf(" the mesh cannot read it back, will not replace it with one of its own, and will not rotate it\n")
|
||||
fmt.Printf(" run `push %s` and `push %s` to send it\n", *provider, node)
|
||||
fmt.Printf(" run `push %s` and `push %s` to send it\n", provider, node)
|
||||
return nil
|
||||
}
|
||||
// A trusted party's secret is announced before it is kept (the confirmation review of 2026-10-09, N1-give):
|
||||
@@ -485,6 +517,22 @@ func secretRotate(ctx context.Context, args []string) error {
|
||||
return errors.New(secretUsage)
|
||||
}
|
||||
node, module, name := rest[0], rest[1], rest[2]
|
||||
return bySwitch(module, secretsOfTheVault,
|
||||
func() error { return rotateKept(ctx, node, module, name, why) },
|
||||
func() error {
|
||||
words, err := rotateInVault(theVault(ctx), ownSecretArgs(node, module, name), actingAs())
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
why.record(ctx, "secret rotate", []string{node, module, name})
|
||||
fmt.Println(words)
|
||||
return nil
|
||||
})
|
||||
}
|
||||
|
||||
// rotateKept is today's `secret rotate`: made anew in the controller's store and sent, for a module whose
|
||||
// secrets are filled the old way.
|
||||
func rotateKept(ctx context.Context, node, module, name string, why handActFlags) error {
|
||||
open, err := openStores(ctx)
|
||||
if err != nil {
|
||||
return err
|
||||
|
||||
@@ -0,0 +1,281 @@
|
||||
package main
|
||||
|
||||
import (
|
||||
"context"
|
||||
"encoding/json"
|
||||
"errors"
|
||||
"fmt"
|
||||
"strings"
|
||||
"time"
|
||||
|
||||
"github.com/nats-io/nats.go"
|
||||
|
||||
"github.com/novox/mesh-controller/internal/broker"
|
||||
"github.com/novox/mesh-controller/internal/link"
|
||||
"github.com/novox/mesh-controller/internal/secrets"
|
||||
)
|
||||
|
||||
// The vault's verbs, reached from the controller's (novox/hq ADR 0349 clauses 8 to 10; design 57's V2, issue 675).
|
||||
//
|
||||
// **Each module is filled the old way or the new way, by its form** (ADR 0349 clause 17, the release lead's
|
||||
// decision of 2026-10-14 on issue 675). `secret accept`, `secret ask` (and `accept --at-desk`), `secret issued`
|
||||
// and `secret rotate` go to the vault only for a module whose secrets are filled the new way — its machine asks
|
||||
// the vault for them — and keep today's path, the controller's store and the declaration it seals, for every other
|
||||
// module. Forwarding a module still filled the old way would report success and change nothing on its machine:
|
||||
// the vault hands it nothing, and its machine never asks.
|
||||
//
|
||||
// **What the controller carries, it cannot open.** The vault takes a value in two steps of one verb: the first
|
||||
// names the secret and answers a key the vault made for that one call; the second carries the value sealed to it.
|
||||
// The value given at this terminal is sealed here before it leaves the process (clause 8); what a desk's prompt or
|
||||
// a provider's tool answers is sealed there to the vault's key, and the controller only passes it on (clauses 9
|
||||
// and 10). The controller still asks the desk and the provider, because the bus lets only the controller ask
|
||||
// them, and asks the vault in its own name, so the vault knows the call is the controller's.
|
||||
|
||||
// secretsOfTheVault says whether a module's secrets are filled the new way: the switch the controller's secret
|
||||
// verbs turn on. **No module's are until design 57's V4** (issue 647), which fills a moved module's
|
||||
// `${secret:}` from the vault and sets this; until then it is false for every module, so every verb keeps today's
|
||||
// path.
|
||||
func secretsOfTheVault(module string) bool { return false }
|
||||
|
||||
// bySwitch runs the vault's path for a module whose secrets are filled the new way, and today's for every other.
|
||||
func bySwitch(module string, ofTheVault func(string) bool, today, vault func() error) error {
|
||||
if ofTheVault != nil && ofTheVault(module) {
|
||||
return vault()
|
||||
}
|
||||
return today()
|
||||
}
|
||||
|
||||
// vaultAsker asks one of the vault's verbs in the controller's name and answers its result, or its refusal.
|
||||
type vaultAsker func(verb string, args map[string]any) (json.RawMessage, error)
|
||||
|
||||
// vaultWithin is how long the vault is given to answer one step.
|
||||
const vaultWithin = 20 * time.Second
|
||||
|
||||
// theVault asks the vault's seat on this controller's bus, on the subject that names the controller.
|
||||
func theVault(ctx context.Context) vaultAsker {
|
||||
return func(verb string, args map[string]any) (json.RawMessage, error) {
|
||||
var result json.RawMessage
|
||||
err := onTheBus(func(conn *nats.Conn) error {
|
||||
answer, err := link.AskMeshSeatToolAs(ctx, conn, broker.ControllerName, "mesh-vault", verb, args, vaultWithin)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
if answer.Error != "" {
|
||||
return errors.New(answer.Error)
|
||||
}
|
||||
result = answer.Result
|
||||
return nil
|
||||
})
|
||||
return result, err
|
||||
}
|
||||
}
|
||||
|
||||
// vaultOpened is a first step's answer: the call, the key made for it, the secret's name in the vault, and who
|
||||
// asked as the vault recorded it. Never a value.
|
||||
type vaultOpened struct {
|
||||
Call string `json:"call"`
|
||||
SealTo string `json:"seal_to"`
|
||||
Secret string `json:"secret"`
|
||||
AskedBy string `json:"asked_by"`
|
||||
}
|
||||
|
||||
// vaultChanged is what the vault kept: never the value.
|
||||
type vaultChanged struct {
|
||||
Name string `json:"name"`
|
||||
Generation int `json:"generation"`
|
||||
By string `json:"by"`
|
||||
Unchanged bool `json:"unchanged"`
|
||||
}
|
||||
|
||||
func openVaultCall(ask vaultAsker, verb string, args map[string]any) (vaultOpened, error) {
|
||||
raw, err := ask(verb, args)
|
||||
if err != nil {
|
||||
return vaultOpened{}, fmt.Errorf("the vault did not open the call: %w", err)
|
||||
}
|
||||
var o vaultOpened
|
||||
if err := json.Unmarshal(raw, &o); err != nil || o.Call == "" || o.SealTo == "" {
|
||||
return vaultOpened{}, errors.New("the vault answered no call and no key to seal to")
|
||||
}
|
||||
return o, nil
|
||||
}
|
||||
|
||||
func endVaultCall(ask vaultAsker, verb string, args map[string]any) (vaultChanged, error) {
|
||||
raw, err := ask(verb, args)
|
||||
if err != nil {
|
||||
return vaultChanged{}, err
|
||||
}
|
||||
var c vaultChanged
|
||||
if err := json.Unmarshal(raw, &c); err != nil {
|
||||
return vaultChanged{}, errors.New("the vault answered something that is not what it kept")
|
||||
}
|
||||
return c, nil
|
||||
}
|
||||
|
||||
// ownSecretArgs names a module's own secret on one machine, as the vault's verbs take it.
|
||||
func ownSecretArgs(node, module, name string) map[string]any {
|
||||
return map[string]any{"node": node, "module": module, "name": name}
|
||||
}
|
||||
|
||||
// provisionArgs names the one credential of a provider assignment and provision.
|
||||
func provisionArgs(provider, provision string) map[string]any {
|
||||
return map[string]any{"provider": provider, "provision": provision}
|
||||
}
|
||||
|
||||
// acceptIntoVault is `secret accept` for a module filled the new way: the value read at this terminal is sealed
|
||||
// here to the key the vault made for this call, and only the sealed box crosses the bus (ADR 0349 clause 8).
|
||||
func acceptIntoVault(ask vaultAsker, target map[string]any, value, by string) (string, error) {
|
||||
first := map[string]any{"by": by}
|
||||
for k, v := range target {
|
||||
first[k] = v
|
||||
}
|
||||
opened, err := openVaultCall(ask, "accept", first)
|
||||
if err != nil {
|
||||
return "", fmt.Errorf("nothing was read or sealed: %w", err)
|
||||
}
|
||||
sealed, err := secrets.Seal(opened.SealTo, []byte(value))
|
||||
value = ""
|
||||
if err != nil {
|
||||
return "", fmt.Errorf("the vault's key for this call is not a sealing key; nothing was sent: %w", err)
|
||||
}
|
||||
changed, err := endVaultCall(ask, "accept", map[string]any{"call": opened.Call, "sealed": sealed})
|
||||
if err != nil {
|
||||
return "", fmt.Errorf("the vault did not take it: %w", err)
|
||||
}
|
||||
if changed.Unchanged {
|
||||
return fmt.Sprintf("the vault already holds this value as %s (generation %d); nothing changed", opened.Secret,
|
||||
changed.Generation), nil
|
||||
}
|
||||
return fmt.Sprintf("the vault now holds %s (generation %d), accepted: sealed at this terminal to the vault's key "+
|
||||
"for this call, opened in the vault alone and never remade by it; its machine takes it from the vault",
|
||||
opened.Secret, changed.Generation), nil
|
||||
}
|
||||
|
||||
// rotateInVault is `secret rotate` for a module filled the new way: the vault makes the value anew and says it
|
||||
// changed; the module's machine takes it from the vault, with no declaration sent.
|
||||
func rotateInVault(ask vaultAsker, target map[string]any, by string) (string, error) {
|
||||
args := map[string]any{"by": by}
|
||||
for k, v := range target {
|
||||
args[k] = v
|
||||
}
|
||||
changed, err := endVaultCall(ask, "rotate", args)
|
||||
if err != nil {
|
||||
return "", fmt.Errorf("not rotated: %w", err)
|
||||
}
|
||||
return fmt.Sprintf("rotated %s in the vault (generation %d), asked by %s; the value is sealed and not shown, and "+
|
||||
"its machine takes it from the vault", changed.Name, changed.Generation, changed.By), nil
|
||||
}
|
||||
|
||||
// giveToVault is the desk's path for a module filled the new way (ADR 0349 clause 9, ADR 0277 §22–§27 in the
|
||||
// vault): the vault records the ask and holds its bounds before the prompt opens and answers the key the prompt
|
||||
// seals to and who asked as it recorded them; the prompt names that asker; what is typed comes back sealed to
|
||||
// the vault's key and is passed on unopened; how the ask ended is the vault's record.
|
||||
func (d deskGive) giveToVault(node, module, name, desk string) (string, error) {
|
||||
opened, err := openVaultCall(d.vault, "secret-ask", map[string]any{"node": node, "module": module, "name": name,
|
||||
"desk": desk, "asked_by": askedByName(d.askedBy)})
|
||||
if err != nil {
|
||||
return "", fmt.Errorf("nobody was asked to type anything: %w", err)
|
||||
}
|
||||
end := func(outcome string) {
|
||||
_, _ = endVaultCall(d.vault, "secret-ask", map[string]any{"call": opened.Call, "outcome": outcome})
|
||||
}
|
||||
raw, err := d.ask(desk, map[string]any{
|
||||
"module": module,
|
||||
"secret": name,
|
||||
"node": node,
|
||||
"asked_by": opened.AskedBy,
|
||||
"seal_to": opened.SealTo,
|
||||
"timeout_seconds": deskPromptWithin,
|
||||
})
|
||||
if err != nil {
|
||||
end("refused")
|
||||
return "", fmt.Errorf("the desk on %s could not be asked: %w", desk, err)
|
||||
}
|
||||
var answer struct {
|
||||
Sealed string `json:"sealed"`
|
||||
Cancelled bool `json:"cancelled"`
|
||||
TimedOut bool `json:"timed_out"`
|
||||
}
|
||||
if err := json.Unmarshal(raw, &answer); err != nil {
|
||||
end("failed")
|
||||
return "", fmt.Errorf("the desk on %s answered something that is not the prompt's answer", desk)
|
||||
}
|
||||
switch {
|
||||
case answer.TimedOut:
|
||||
end("timed-out")
|
||||
return "", fmt.Errorf("%w: the prompt on %s was not answered within %d seconds", errNothingGiven, desk, deskPromptWithin)
|
||||
case answer.Cancelled:
|
||||
end("dismissed")
|
||||
return "", fmt.Errorf("%w: the prompt on %s was dismissed", errNothingGiven, desk)
|
||||
case answer.Sealed == "":
|
||||
end("failed")
|
||||
return "", fmt.Errorf("the desk on %s answered no sealed value", desk)
|
||||
}
|
||||
changed, err := endVaultCall(d.vault, "secret-ask", map[string]any{"call": opened.Call, "sealed": answer.Sealed})
|
||||
if err != nil {
|
||||
return "", fmt.Errorf("the vault did not take what was typed at the desk on %s: %w", desk, err)
|
||||
}
|
||||
act := link.HandAct{Verb: "secret accept", Args: []string{node, module, name, "--at-desk", desk},
|
||||
Why: fmt.Sprintf("the operator gave %s for %s on %s at the desk on %s, asked by %s; kept in the vault", name,
|
||||
module, node, desk, opened.AskedBy),
|
||||
Cause: "given-at-the-desk"}
|
||||
recorded := ""
|
||||
if err := d.record(act); err != nil {
|
||||
recorded = fmt.Sprintf("\n this act could NOT be recorded in the hand-act log, and is done anyway: %v", err)
|
||||
}
|
||||
if d.announce != nil {
|
||||
if err := d.announce(node, module, name, "at the desk on "+desk); err != nil {
|
||||
recorded += fmt.Sprintf("\n this change could NOT be announced on the operator's channels: %v", err)
|
||||
}
|
||||
}
|
||||
return fmt.Sprintf("%s on %s now holds %q, given at the desk on %s and kept in the vault as %s (generation %d); "+
|
||||
"it opened in the vault alone, and %s takes it from there", module, node, name, desk, opened.Secret,
|
||||
changed.Generation, node) + recorded, nil
|
||||
}
|
||||
|
||||
// takeIntoVault is `secret issued` for a provider filled the new way (ADR 0349 clause 10): the vault opens a call;
|
||||
// the provider's tool, asked by the controller as today — the one caller the bus lets ask it — seals its
|
||||
// software's value to the vault's key; the controller passes it on unopened. Nothing is sealed or sent here: each
|
||||
// holder takes it from the vault.
|
||||
func (t issuedTaker) takeIntoVault(provision string, p issuedProvider) (string, error) {
|
||||
opened, err := openVaultCall(t.vault, "issued", map[string]any{"provision": provision, "provider": p.Node,
|
||||
"module": p.Module})
|
||||
if err != nil {
|
||||
return "", err
|
||||
}
|
||||
answer, err := t.ask(p, map[string]any{"provision": provision, "seal_to": opened.SealTo})
|
||||
if err != nil {
|
||||
return "", fmt.Errorf("its tool %s could not be asked: %w", p.Tool, err)
|
||||
}
|
||||
if answer.Error != "" {
|
||||
return "", fmt.Errorf("its tool %s refused: %s", p.Tool, answer.Error)
|
||||
}
|
||||
var given struct {
|
||||
Sealed string `json:"sealed"`
|
||||
}
|
||||
if err := unmarshalAnswer(answer, &given); err != nil || given.Sealed == "" {
|
||||
return "", fmt.Errorf("its tool %s answered no sealed value", p.Tool)
|
||||
}
|
||||
changed, err := endVaultCall(t.vault, "issued", map[string]any{"call": opened.Call, "sealed": given.Sealed})
|
||||
if err != nil {
|
||||
return "", fmt.Errorf("the vault did not take what its tool %s answered: %w", p.Tool, err)
|
||||
}
|
||||
if changed.Unchanged {
|
||||
return fmt.Sprintf("%s from %s on %s: the vault already holds the value its software issued (generation %d); "+
|
||||
"nothing was written", provision, p.Module, p.Node, changed.Generation), nil
|
||||
}
|
||||
recorded := ""
|
||||
if err := t.record(link.HandAct{Verb: handActIssued, Args: []string{provision},
|
||||
Why: fmt.Sprintf("%s on %s gave the credential its software issued for %s; kept in the vault as %s", p.Module,
|
||||
p.Node, provision, opened.Secret),
|
||||
Cause: "issued-by-software"}); err != nil {
|
||||
recorded = fmt.Sprintf("\n this was NOT recorded in the hand-act log, and is done anyway: %v", err)
|
||||
}
|
||||
return fmt.Sprintf("%s from %s on %s: the value its software issued is kept in the vault as %s (generation %d); "+
|
||||
"each holder takes it from there%s", provision, p.Module, p.Node, opened.Secret, changed.Generation,
|
||||
recorded), nil
|
||||
}
|
||||
|
||||
// actingAs is who acts in this process, as the vault records it beside the controller that carries the act.
|
||||
func actingAs() string {
|
||||
return strings.TrimSpace(askedByName(link.Caller()))
|
||||
}
|
||||
@@ -0,0 +1,346 @@
|
||||
package main
|
||||
|
||||
import (
|
||||
"bytes"
|
||||
"context"
|
||||
"encoding/json"
|
||||
"errors"
|
||||
"fmt"
|
||||
"log"
|
||||
"strings"
|
||||
"sync"
|
||||
"testing"
|
||||
|
||||
"github.com/nats-io/nats.go"
|
||||
|
||||
"github.com/novox/mesh-controller/internal/broker"
|
||||
"github.com/novox/mesh-controller/internal/inventory"
|
||||
"github.com/novox/mesh-controller/internal/link"
|
||||
"github.com/novox/mesh-controller/internal/secrets"
|
||||
"github.com/novox/mesh-controller/internal/testbus"
|
||||
)
|
||||
|
||||
// The controller's secret verbs reach the vault only for a module filled the new way (design 57's V2, issue 675,
|
||||
// the release lead's decision on it; novox/hq ADR 0349 clauses 8 to 10 and 17).
|
||||
|
||||
// standInVault answers the vault's two-step verbs as mesh-vault does: a key made for each call, the value opened
|
||||
// with it here and nowhere else. It keeps what it took and every argument it was sent, as the bus carried them.
|
||||
type standInVault struct {
|
||||
t *testing.T
|
||||
mu sync.Mutex
|
||||
keys map[string]string // call → private key
|
||||
taken map[string]string // secret → value
|
||||
sent []string // every argument, as JSON
|
||||
refuse error // refuses every first step when set
|
||||
outcome []string
|
||||
again bool // the value is held already
|
||||
}
|
||||
|
||||
func newStandInVault(t *testing.T) *standInVault {
|
||||
return &standInVault{t: t, keys: map[string]string{}, taken: map[string]string{}}
|
||||
}
|
||||
|
||||
func (v *standInVault) ask(verb string, args map[string]any) (json.RawMessage, error) {
|
||||
v.mu.Lock()
|
||||
defer v.mu.Unlock()
|
||||
raw, _ := json.Marshal(map[string]any{"verb": verb, "args": args})
|
||||
v.sent = append(v.sent, string(raw))
|
||||
if verb == "rotate" {
|
||||
return json.Marshal(vaultChanged{Name: "own/g14/dunst/token", Generation: 2, By: fmt.Sprint(args["by"])})
|
||||
}
|
||||
call, _ := args["call"].(string)
|
||||
if call == "" {
|
||||
if v.refuse != nil {
|
||||
return nil, v.refuse
|
||||
}
|
||||
public, private, err := secrets.Keypair()
|
||||
if err != nil {
|
||||
v.t.Fatal(err)
|
||||
}
|
||||
call = fmt.Sprintf("call-%d", len(v.keys)+1)
|
||||
v.keys[call] = private
|
||||
secret := fmt.Sprintf("%s/%v", verb, args["name"])
|
||||
return json.Marshal(vaultOpened{Call: call, SealTo: public, Secret: secret, AskedBy: "as-the-vault-recorded"})
|
||||
}
|
||||
private, ok := v.keys[call]
|
||||
if !ok {
|
||||
return nil, errors.New("no such call")
|
||||
}
|
||||
delete(v.keys, call)
|
||||
if o, _ := args["outcome"].(string); o != "" {
|
||||
v.outcome = append(v.outcome, o)
|
||||
return json.Marshal(vaultChanged{Unchanged: true})
|
||||
}
|
||||
value, err := secrets.Open(private, fmt.Sprint(args["sealed"]))
|
||||
if err != nil {
|
||||
return nil, errors.New("not sealed to this call")
|
||||
}
|
||||
// As mesh-vault takes a typed value: its line ending off (intake.go AskEnd).
|
||||
v.taken[verb] = strings.TrimRight(string(value), "\r\n")
|
||||
return json.Marshal(vaultChanged{Name: verb, Generation: 1, Unchanged: v.again})
|
||||
}
|
||||
|
||||
func (v *standInVault) said() string {
|
||||
v.mu.Lock()
|
||||
defer v.mu.Unlock()
|
||||
return strings.Join(v.sent, "\n")
|
||||
}
|
||||
|
||||
// No module's secrets are filled the new way before V4: every verb keeps today's path for every module.
|
||||
func TestNoModulesSecretsAreTheVaultsBeforeV4(t *testing.T) {
|
||||
for _, module := range []string{"telegram", "dunst", "plex", "postgres", "mesh-vault", "kometa", ""} {
|
||||
if secretsOfTheVault(module) {
|
||||
t.Errorf("%q is filled the new way before V4", module)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// The switch: today's path for a module filled the old way, the vault's for one filled the new way, and never
|
||||
// both.
|
||||
func TestTheSwitchRunsOnePathByTheModulesForm(t *testing.T) {
|
||||
moved := func(m string) bool { return m == "dunst" }
|
||||
for module, want := range map[string]string{"telegram": "today", "dunst": "vault"} {
|
||||
var ran []string
|
||||
err := bySwitch(module, moved,
|
||||
func() error { ran = append(ran, "today"); return nil },
|
||||
func() error { ran = append(ran, "vault"); return nil })
|
||||
if err != nil || strings.Join(ran, ",") != want {
|
||||
t.Errorf("%s ran %v (%v), not %s", module, ran, err, want)
|
||||
}
|
||||
}
|
||||
var ran []string
|
||||
_ = bySwitch("dunst", nil, func() error { ran = append(ran, "today"); return nil },
|
||||
func() error { ran = append(ran, "vault"); return nil })
|
||||
if strings.Join(ran, ",") != "today" {
|
||||
t.Errorf("no switch ran %v", ran)
|
||||
}
|
||||
}
|
||||
|
||||
// `secret accept` for a module filled the new way: the value read at the terminal is sealed there to the key the
|
||||
// vault made for this one call; no clear value crosses the bus or reaches the controller's log. Over a real bus,
|
||||
// with a tap on every subject.
|
||||
func TestAnAcceptedValueCrossesTheBusOnlySealedToTheVault(t *testing.T) {
|
||||
conn, err := nats.Connect(testbus.URL(t))
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
t.Cleanup(conn.Close)
|
||||
before := handActConn
|
||||
handActConn = conn
|
||||
t.Cleanup(func() { handActConn = before })
|
||||
|
||||
var tapped bytes.Buffer
|
||||
var tapMu sync.Mutex
|
||||
tap, err := conn.Subscribe(">", func(m *nats.Msg) {
|
||||
tapMu.Lock()
|
||||
defer tapMu.Unlock()
|
||||
tapped.WriteString(m.Subject + " " + string(m.Data) + "\n")
|
||||
})
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
defer tap.Unsubscribe()
|
||||
|
||||
vault := newStandInVault(t)
|
||||
var subjects []string
|
||||
holder, err := conn.Subscribe(broker.CalledPattern(link.SeatToolSubject("mesh-vault", "accept")), func(m *nats.Msg) {
|
||||
subjects = append(subjects, m.Subject)
|
||||
var args map[string]any
|
||||
_ = json.Unmarshal(m.Data, &args)
|
||||
result, err := vault.ask("accept", args)
|
||||
answer := link.Answer{Result: result}
|
||||
if err != nil {
|
||||
answer = link.Answer{Error: err.Error()}
|
||||
}
|
||||
raw, _ := json.Marshal(answer)
|
||||
_ = m.Respond(raw)
|
||||
})
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
defer holder.Unsubscribe()
|
||||
|
||||
var logged bytes.Buffer
|
||||
log.SetOutput(&logged)
|
||||
t.Cleanup(func() { log.SetOutput(nil) })
|
||||
|
||||
const value = "pw-of-last-year-that-must-not-show"
|
||||
words, err := acceptIntoVault(theVault(context.Background()), provisionArgs("novox", "postgres"), value, "jochen")
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if vault.taken["accept"] != value {
|
||||
t.Fatalf("the vault opened %q", vault.taken["accept"])
|
||||
}
|
||||
if err := conn.Flush(); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
tapMu.Lock()
|
||||
crossed := tapped.String()
|
||||
tapMu.Unlock()
|
||||
if !strings.Contains(crossed, "mesh.seat.mesh-vault.call.accept.controller") {
|
||||
t.Fatalf("the tap heard no call to the vault: %s", crossed)
|
||||
}
|
||||
for _, s := range subjects {
|
||||
if !strings.HasSuffix(s, ".controller") {
|
||||
t.Errorf("the vault was asked on %s, not in the controller's name", s)
|
||||
}
|
||||
}
|
||||
for where, text := range map[string]string{"the bus": crossed, "the log": logged.String(), "the answer": words} {
|
||||
if strings.Contains(text, value) || strings.Contains(text, "must-not-show") {
|
||||
t.Errorf("the value crossed %s in the clear", where)
|
||||
}
|
||||
}
|
||||
if !strings.Contains(words, "sealed at this terminal") {
|
||||
t.Errorf("the answer: %q", words)
|
||||
}
|
||||
// A vault that refuses the first step: nothing is sealed or sent.
|
||||
vault.refuse = errors.New("the vault has no operator key")
|
||||
if _, err := acceptIntoVault(vault.ask, ownSecretArgs("g14", "dunst", "token"), value, "jochen"); err == nil ||
|
||||
!strings.Contains(err.Error(), "nothing was read or sealed") {
|
||||
t.Errorf("a refused call said %v", err)
|
||||
}
|
||||
}
|
||||
|
||||
// The desk's path for a module filled the new way: the vault records the ask and holds its bounds before the
|
||||
// prompt; the prompt names the asker as the vault recorded it; what is typed passes the controller sealed and
|
||||
// unopened; today's store is not touched. For a module filled the old way, the vault is never asked.
|
||||
func TestAnAskForAModuleFilledTheNewWayIsTheVaults(t *testing.T) {
|
||||
d, accepted, acts, asked := aDesk(t, sealedTo(t, typed))
|
||||
d.askedBy = "g14/claude-code"
|
||||
d.open = func(node, module, name, desk string) (int64, error) {
|
||||
t.Fatal("today's record was asked")
|
||||
return 0, nil
|
||||
}
|
||||
vault := newStandInVault(t)
|
||||
d.vault = vault.ask
|
||||
words, err := d.give("anchor", "telegram", "telegram-token", "laptop")
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if len(*accepted) != 0 {
|
||||
t.Fatal("the controller opened and kept what was typed")
|
||||
}
|
||||
if vault.taken["secret-ask"] != typed {
|
||||
t.Fatalf("the vault opened %q", vault.taken["secret-ask"])
|
||||
}
|
||||
if p := (*asked)[0]; p["asked_by"] != "as-the-vault-recorded" || p["seal_to"] == "" {
|
||||
t.Errorf("the prompt was asked %v", p)
|
||||
}
|
||||
if !strings.Contains(vault.said(), `"asked_by":"g14/claude-code"`) {
|
||||
t.Errorf("the vault was not told who asked: %s", vault.said())
|
||||
}
|
||||
if len(*acts) != 1 || (*acts)[0].Cause != "given-at-the-desk" || !strings.Contains((*acts)[0].Why, "kept in the vault") {
|
||||
t.Errorf("the act: %+v", *acts)
|
||||
}
|
||||
raw, _ := json.Marshal(struct {
|
||||
Words string
|
||||
Acts []link.HandAct
|
||||
Asked []map[string]any
|
||||
}{words, *acts, *asked})
|
||||
if strings.Contains(string(raw)+vault.said(), typed) || strings.Contains(string(raw)+vault.said(), "AAEhBP0") {
|
||||
t.Fatal("the value appears in what was said, asked, recorded or sent to the vault")
|
||||
}
|
||||
|
||||
// Dismissed: the vault's record ends the ask so.
|
||||
d2, _, _, _ := aDesk(t, func(map[string]any) (json.RawMessage, error) { return json.RawMessage(`{"cancelled":true}`), nil })
|
||||
v2 := newStandInVault(t)
|
||||
d2.vault = v2.ask
|
||||
if _, err := d2.give("anchor", "telegram", "telegram-token", "laptop"); !errors.Is(err, errNothingGiven) {
|
||||
t.Fatalf("%v", err)
|
||||
}
|
||||
if strings.Join(v2.outcome, ",") != "dismissed" {
|
||||
t.Errorf("the vault was told %v", v2.outcome)
|
||||
}
|
||||
// The vault's bounds refuse: nobody is asked to type.
|
||||
d3, _, _, asked3 := aDesk(t, sealedTo(t, typed))
|
||||
v3 := newStandInVault(t)
|
||||
v3.refuse = errors.New("an ask for telegram-token of telegram on anchor is still open")
|
||||
d3.vault = v3.ask
|
||||
if _, err := d3.give("anchor", "telegram", "telegram-token", "laptop"); err == nil ||
|
||||
!strings.Contains(err.Error(), "nobody was asked to type anything") || len(*asked3) != 0 {
|
||||
t.Errorf("a refused ask: %v, prompts %d", err, len(*asked3))
|
||||
}
|
||||
|
||||
// Filled the old way: today's path, the vault never asked.
|
||||
d4, accepted4, _, _ := aDesk(t, sealedTo(t, typed))
|
||||
if _, err := d4.give("anchor", "telegram", "telegram-token", "laptop"); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if len(*accepted4) != 1 || d4.vault != nil {
|
||||
t.Error("a module filled the old way did not take today's path")
|
||||
}
|
||||
}
|
||||
|
||||
// `secret issued` for a provider filled the new way: the provider's tool seals to the vault's key for the call,
|
||||
// the controller passes it on unopened, and nothing is kept, sealed or sent here. A provider filled the old way
|
||||
// keeps today's path and the vault is never asked.
|
||||
func TestAnIssuedValueForAProviderFilledTheNewWayGoesToTheVault(t *testing.T) {
|
||||
var r issuedRecorder
|
||||
vault := newStandInVault(t)
|
||||
taker := r.taker(t, issuedSealed(t, "the-servers-own-token"))
|
||||
taker.keep = func(string, string, string) (inventory.IssuedKept, error) {
|
||||
t.Fatal("today's store was asked to keep it")
|
||||
return inventory.IssuedKept{}, nil
|
||||
}
|
||||
taker.ofTheVault = func(m string) bool { return m == "plex" }
|
||||
taker.vault = vault.ask
|
||||
words, err := taker.take("plex-api")
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if vault.taken["issued"] != "the-servers-own-token" {
|
||||
t.Fatalf("the vault opened %q", vault.taken["issued"])
|
||||
}
|
||||
if len(r.sent) != 0 {
|
||||
t.Errorf("machines were sent: %v", r.sent)
|
||||
}
|
||||
if len(r.acts) != 1 || r.acts[0].Verb != handActIssued || !strings.Contains(r.acts[0].Why, "kept in the vault") {
|
||||
t.Errorf("the act: %+v", r.acts)
|
||||
}
|
||||
if !strings.Contains(vault.said(), `"provider":"ace"`) || !strings.Contains(vault.said(), `"module":"plex"`) {
|
||||
t.Errorf("the vault was told %s", vault.said())
|
||||
}
|
||||
for _, said := range []string{words, r.acts[0].Why, vault.said()} {
|
||||
if strings.Contains(said, "the-servers-own-token") {
|
||||
t.Fatalf("the value was said: %q", said)
|
||||
}
|
||||
}
|
||||
// The same value again: the vault writes nothing, and nothing is recorded.
|
||||
again := newStandInVault(t)
|
||||
again.again = true
|
||||
r2 := issuedRecorder{}
|
||||
t2 := r2.taker(t, issuedSealed(t, "the-servers-own-token"))
|
||||
t2.ofTheVault, t2.vault = func(string) bool { return true }, again.ask
|
||||
if words, err := t2.take("plex-api"); err != nil || len(r2.acts) != 0 || !strings.Contains(words, "nothing was written") {
|
||||
t.Errorf("%q %v %+v", words, err, r2.acts)
|
||||
}
|
||||
|
||||
// Filled the old way: today's path.
|
||||
var r3 issuedRecorder
|
||||
t3 := r3.taker(t, issuedSealed(t, "the-servers-own-token"))
|
||||
t3.ofTheVault = func(string) bool { return false }
|
||||
t3.vault = func(string, map[string]any) (json.RawMessage, error) {
|
||||
t.Fatal("the vault was asked for a provider filled the old way")
|
||||
return nil, nil
|
||||
}
|
||||
if _, err := t3.take("plex-api"); err != nil || len(r3.kept) != 1 {
|
||||
t.Errorf("today's path: %v, kept %d", err, len(r3.kept))
|
||||
}
|
||||
}
|
||||
|
||||
// `secret rotate` for a module filled the new way asks the vault, naming who asked; nothing is sent.
|
||||
func TestARotationForAModuleFilledTheNewWayIsTheVaults(t *testing.T) {
|
||||
vault := newStandInVault(t)
|
||||
words, err := rotateInVault(vault.ask, ownSecretArgs("g14", "dunst", "token"), "g14/claude-code")
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if !strings.Contains(vault.said(), `"verb":"rotate"`) || !strings.Contains(vault.said(), `"by":"g14/claude-code"`) ||
|
||||
!strings.Contains(vault.said(), `"module":"dunst"`) {
|
||||
t.Errorf("the vault was told %s", vault.said())
|
||||
}
|
||||
if !strings.Contains(words, "generation 2") || !strings.Contains(words, "takes it from the vault") {
|
||||
t.Errorf("%q", words)
|
||||
}
|
||||
}
|
||||
@@ -40,39 +40,6 @@ var (
|
||||
AskedBucket = BucketName(ControllerSeat, "asked")
|
||||
)
|
||||
|
||||
// The controller's record mirrors (novox/hq ADR 0289, design 54, design 57 package F3): a bucket per kind of
|
||||
// record, one key per record, written from the store by the lease holder only when the record changed, with
|
||||
// one event on the controller's seat per write. Read by whatever names them in `reads` as
|
||||
// `mesh-controller.<mirror>` (ADR 0201 §1), written by nobody else. **No secret in either** (ADR 0201): an
|
||||
// assignment's value names its module and node; a module's declared config is what its manifest, which is
|
||||
// public, declares.
|
||||
//
|
||||
// AssignmentsBucket holds which module is assigned to which node, keyed `<node>.<module>`; and
|
||||
// DeclaredConfigBucket each registered module's config declarations as its manifest holds them, keyed by
|
||||
// the module: what mesh-config resolves a config value from (ADR 0348 clauses 2 and 4).
|
||||
var (
|
||||
AssignmentsBucket = BucketName(ControllerSeat, MirrorAssignments)
|
||||
DeclaredConfigBucket = BucketName(ControllerSeat, MirrorDeclaredConfig)
|
||||
)
|
||||
|
||||
// The mirrors' local names, as a reader names them in `reads` after `mesh-controller.`.
|
||||
const (
|
||||
MirrorAssignments = "assignments"
|
||||
MirrorDeclaredConfig = "declared-config"
|
||||
)
|
||||
|
||||
// ControllerMirrors are the local names of the controller's record mirrors, which a module may read.
|
||||
func ControllerMirrors() []string { return []string{MirrorAssignments, MirrorDeclaredConfig} }
|
||||
|
||||
// The mirrors' caps (design 54 §5): one value per key, no age, 64 KiB a value, 64 MiB a bucket.
|
||||
const (
|
||||
MirrorValueBytes = 64 << 10
|
||||
MirrorBytes = 64 << 20
|
||||
// MirrorMarkerTTL is the longest a purge's marker is kept (design 54 §2): the bucket allows a
|
||||
// per-message age for it, and the writer gives each marker a day.
|
||||
MirrorMarkerTTL = 24 * time.Hour
|
||||
)
|
||||
|
||||
// AskedKeptFor is how long an ask is kept after it was made: a month, as the router keeps its own.
|
||||
const AskedKeptFor = 30 * 24 * time.Hour
|
||||
|
||||
@@ -99,14 +66,12 @@ const (
|
||||
// IsControllerBucket says a bucket is the controller's own, not a module's state nothing declares.
|
||||
func IsControllerBucket(bucket string) bool {
|
||||
return bucket == CallsBucket || bucket == HandActsBucket || bucket == ConditionsBucket ||
|
||||
bucket == ConditionHistoryBucket || bucket == LeaseBucket || bucket == AskedBucket ||
|
||||
bucket == AssignmentsBucket || bucket == DeclaredConfigBucket
|
||||
bucket == ConditionHistoryBucket || bucket == LeaseBucket || bucket == AskedBucket
|
||||
}
|
||||
|
||||
// ControllerBuckets are the controller's own buckets, in the order they are asserted.
|
||||
func ControllerBuckets() []string {
|
||||
return []string{LeaseBucket, CallsBucket, HandActsBucket, ConditionsBucket, ConditionHistoryBucket, AskedBucket,
|
||||
AssignmentsBucket, DeclaredConfigBucket}
|
||||
return []string{LeaseBucket, CallsBucket, HandActsBucket, ConditionsBucket, ConditionHistoryBucket, AskedBucket}
|
||||
}
|
||||
|
||||
// ControllerBucketsAsserter is what raising the controller's buckets needs of a connection.
|
||||
@@ -203,27 +168,6 @@ func (j *JetStream) EnsureControllerBuckets() error {
|
||||
}); err != nil {
|
||||
return fmt.Errorf("asserting bucket %s: %w", AskedBucket, err)
|
||||
}
|
||||
// **The record mirrors: no age** (design 54 §5). A key is removed when its record is, by a purge whose
|
||||
// marker lives a day; a key that expired would be a record the bus forgot while the store still holds it.
|
||||
for _, m := range []struct{ bucket, why string }{
|
||||
{AssignmentsBucket, "which module is assigned to which node, keyed <node>.<module> (novox/hq ADR 0289, " +
|
||||
"design 54): written by the controller alone from its store when an assignment changes, read through `reads`"},
|
||||
{DeclaredConfigBucket, "each registered module's config declarations as its manifest holds them, keyed by " +
|
||||
"the module (novox/hq ADR 0348, design 57 F3): written by the controller alone from its store when a " +
|
||||
"registration changes them, read through `reads`"},
|
||||
} {
|
||||
if _, err := js.CreateOrUpdateKeyValue(ctx, jetstream.KeyValueConfig{
|
||||
Bucket: m.bucket,
|
||||
Description: m.why,
|
||||
History: 1,
|
||||
MaxValueSize: MirrorValueBytes,
|
||||
MaxBytes: MirrorBytes,
|
||||
Storage: jetstream.FileStorage,
|
||||
LimitMarkerTTL: MirrorMarkerTTL,
|
||||
}); err != nil {
|
||||
return fmt.Errorf("asserting bucket %s: %w", m.bucket, err)
|
||||
}
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
|
||||
@@ -21,8 +21,7 @@ func TestTheControllerMayWriteEveryBucketItWrites(t *testing.T) {
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
want := []string{"$KV." + CallsBucket + ".>", "$KV." + HandActsBucket + ".>",
|
||||
"$KV." + AssignmentsBucket + ".>", "$KV." + DeclaredConfigBucket + ".>"}
|
||||
want := []string{"$KV." + CallsBucket + ".>", "$KV." + HandActsBucket + ".>"}
|
||||
for _, seat := range seatsTheControllerAsks {
|
||||
if hasCancelledSet(seat) {
|
||||
want = append(want, "$KV."+CancelledSetName(seat)+".>")
|
||||
@@ -38,34 +37,6 @@ func TestTheControllerMayWriteEveryBucketItWrites(t *testing.T) {
|
||||
}
|
||||
}
|
||||
|
||||
// **A module reads a record mirror through `reads`, and never writes it** (ADR 0201 §1, design 54 §5,
|
||||
// design 57 F3): granted the direct reads and watches of the bucket, and refused at composition a grant to
|
||||
// write it, whose writer is the controller.
|
||||
func TestAModuleReadsTheControllersMirrorsAndWritesNone(t *testing.T) {
|
||||
reader := Principal{Kind: KindModule, Node: "one", Module: "mesh-config", PasswordHash: "x",
|
||||
Reads: []string{"mesh-controller." + MirrorAssignments, "mesh-controller." + MirrorDeclaredConfig}}
|
||||
p, err := PermissionsFor(reader)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
for _, bucket := range []string{AssignmentsBucket, DeclaredConfigBucket} {
|
||||
for _, subject := range []string{"$JS.API.STREAM.INFO.KV_" + bucket, "$JS.API.DIRECT.GET.KV_" + bucket + ".>",
|
||||
"$JS.API.CONSUMER.CREATE.KV_" + bucket + ".>"} {
|
||||
if !slices.Contains(p.Publish, subject) {
|
||||
t.Errorf("a module reading %s may not publish %s", bucket, subject)
|
||||
}
|
||||
}
|
||||
for _, subject := range p.Publish {
|
||||
if SubjectsOverlap(subject, "$KV."+bucket+".>") {
|
||||
t.Errorf("a module reading %s may write it through %s", bucket, subject)
|
||||
}
|
||||
}
|
||||
if err := CheckWriters(reader, []string{"$KV." + bucket + ".>"}); err == nil {
|
||||
t.Errorf("a module granted a write of %s would compose", bucket)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// **A machine's node tools may say they are there, as that machine and no other** (novox/hq to-be 45
|
||||
// S11), and the controller may hear the bus's advisories and ask who answers — read-only, named.
|
||||
func TestTheWatchedSignalsMayBeSaidAndHeard(t *testing.T) {
|
||||
@@ -126,55 +97,46 @@ func TestWhatTheControllerAskedIsBounded(t *testing.T) {
|
||||
}
|
||||
}
|
||||
|
||||
// **A listed mirror is no undeclared bucket** (ADR 0356 §3, ADR 0201 §16): the controller's list of its mirrors
|
||||
// stands for their declaration, so the report of state nothing declares never names one.
|
||||
func TestAListedMirrorIsNotReportedAsUndeclared(t *testing.T) {
|
||||
names := []string{"gone_state"}
|
||||
for _, m := range ControllerMirrors() {
|
||||
names = append(names, BucketName(ControllerSeat, m))
|
||||
}
|
||||
undeclared, err := RaiseBuckets(fakeBuckets{names: names}, nil)
|
||||
// **The controller may ask the vault's intake verbs in its own name** (design 57's V2, issue 675): the vault
|
||||
// takes a value only from the controller, which it knows from the subject that names its caller — so the
|
||||
// controller is granted that subject for the verbs the vault's holder serves, and no more of the vault's seat.
|
||||
// Composed from the records, as the issued tools are: the installer's first user list carries none.
|
||||
func TestTheControllerMayAskTheVaultInItsOwnName(t *testing.T) {
|
||||
vault := Declared{Module: "mesh-vault", Holds: []Seat{{Name: "mesh-vault",
|
||||
Serves: []string{"get", "accept", "secret-ask", "issued", "rotate"}}}}
|
||||
users, err := Users(Records{Nodes: []string{"novox"}, Assigned: map[string][]Declared{"novox": {vault}}})
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if !slices.Equal(undeclared, []string{"gone_state"}) {
|
||||
t.Fatalf("the report of undeclared state names %v", undeclared)
|
||||
controller := users[0]
|
||||
if strings.Join(controller.AsksTheVault, ",") != "accept,issued,rotate,secret-ask" {
|
||||
t.Fatalf("composed %v", controller.AsksTheVault)
|
||||
}
|
||||
}
|
||||
|
||||
// **The mirrors' list and the writers table agree** (ADR 0356 §3, to-be 45 §1): every listed mirror is a row of
|
||||
// the table whose one writer is the controller, on its bucket alone; and every row that writes a bucket of the
|
||||
// controller's named as a mirror is on the list.
|
||||
func TestTheMirrorsListAndTheWritersTableAgree(t *testing.T) {
|
||||
controller := Principal{Kind: KindController}
|
||||
module := Principal{Kind: KindModule, Module: ControllerSeat, Node: "one"}
|
||||
rowed := map[string]bool{}
|
||||
for _, row := range WritersTable {
|
||||
if !strings.HasSuffix(row.State, " mirror") {
|
||||
continue
|
||||
p, err := PermissionsFor(controller)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
for _, verb := range []string{"accept", "secret-ask", "issued", "rotate"} {
|
||||
if !MayPublish(p, CalledSubject("mesh.seat.mesh-vault.tool."+verb, ControllerName)) {
|
||||
t.Errorf("the controller may not ask the vault's %s in its own name", verb)
|
||||
}
|
||||
for _, subject := range row.Subjects {
|
||||
bucket := strings.TrimSuffix(strings.TrimPrefix(subject, "$KV."), ".>")
|
||||
rowed[bucket] = true
|
||||
if row.Writes == nil || !row.Writes(controller, subject) || row.Writes(module, subject) {
|
||||
t.Errorf("%q is not written by the controller alone", row.State)
|
||||
}
|
||||
if MayPublish(p, CalledSubject("mesh.seat.mesh-vault.tool."+verb, "anchor.node-tools")) {
|
||||
t.Errorf("the controller may ask the vault's %s in another's name", verb)
|
||||
}
|
||||
}
|
||||
listed := map[string]bool{}
|
||||
for _, m := range ControllerMirrors() {
|
||||
bucket := BucketName(ControllerSeat, m)
|
||||
listed[bucket] = true
|
||||
if !rowed[bucket] {
|
||||
t.Errorf("the mirror %s is listed and no row of the writers table writes it", bucket)
|
||||
}
|
||||
if !IsControllerBucket(bucket) || !slices.Contains(ControllerBuckets(), bucket) {
|
||||
t.Errorf("the mirror %s is listed and not asserted as the controller's own", bucket)
|
||||
}
|
||||
if MayPublish(p, CalledSubject("mesh.seat.mesh-vault.tool.get", ControllerName)) {
|
||||
t.Error("the controller may ask the vault for a secret: it seals nothing the vault holds")
|
||||
}
|
||||
for bucket := range rowed {
|
||||
if !listed[bucket] {
|
||||
t.Errorf("the writers table writes the mirror %s, which the list does not name", bucket)
|
||||
// A vault that serves none of them yet — the running one — composes no grant, nor does genesis.
|
||||
older := Declared{Module: "mesh-vault", Holds: []Seat{{Name: "mesh-vault", Serves: []string{"get"}}}}
|
||||
users, _ = Users(Records{Nodes: []string{"novox"}, Assigned: map[string][]Declared{"novox": {older}}})
|
||||
if len(users[0].AsksTheVault) != 0 {
|
||||
t.Errorf("a vault serving none composed %v", users[0].AsksTheVault)
|
||||
}
|
||||
genesis, _ := PermissionsFor(Principal{Kind: KindController, PasswordHash: "x"})
|
||||
for _, s := range genesis.Publish {
|
||||
if strings.Contains(s, "mesh-vault") {
|
||||
t.Errorf("the installer's controller may publish %s", s)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
@@ -172,9 +172,7 @@ func TestRaisingLogsReportsWhatNothingDeclares(t *testing.T) {
|
||||
// **No path of the mesh deletes or purges a log or a bucket, or recreates one** (novox/hq ADR 0297 §2,
|
||||
// ADR 0201 §15–16): no code of this repository outside its tests calls the client's stream or bucket
|
||||
// delete, or purges a stream, but for the controller lease's own stream, which purges its own history
|
||||
// below a sequence (internal/lease), and the controller's record mirrors, which purge the key of a record
|
||||
// the store no longer holds (internal/recordmirror; novox/hq ADR 0289, design 54 §2). A new call is a decision,
|
||||
// not a refactor.
|
||||
// below a sequence (internal/lease). A new call is a decision, not a refactor.
|
||||
func TestNoPathDeletesALogOrABucket(t *testing.T) {
|
||||
removers := map[string]bool{"DeleteStream": true, "DeleteKeyValue": true, "PurgeStream": true,
|
||||
"DeleteObjectStore": true}
|
||||
@@ -199,9 +197,7 @@ func TestNoPathDeletesALogOrABucket(t *testing.T) {
|
||||
return true
|
||||
}
|
||||
name := sel.Sel.Name
|
||||
decided := strings.Contains(filepath.ToSlash(path), "internal/lease/") ||
|
||||
strings.Contains(filepath.ToSlash(path), "internal/recordmirror/")
|
||||
if removers[name] || (name == "Purge" && !decided) {
|
||||
if removers[name] || (name == "Purge" && !strings.Contains(filepath.ToSlash(path), "internal/lease/")) {
|
||||
found = append(found, path+": "+name)
|
||||
}
|
||||
return true
|
||||
|
||||
@@ -16,6 +16,7 @@ import (
|
||||
"errors"
|
||||
"fmt"
|
||||
"regexp"
|
||||
"slices"
|
||||
"sort"
|
||||
"strings"
|
||||
"time"
|
||||
@@ -130,6 +131,11 @@ type Principal struct {
|
||||
// grants.
|
||||
AsksIssued []string
|
||||
|
||||
// AsksTheVault are the vault's verbs the controller asks in its own name (VerbsTheControllerAsksTheVault,
|
||||
// issue 675), those the vault's holder serves: on the subject that names its caller, so the vault knows the
|
||||
// controller asks. Composed from the records like AsksIssued, so the installer's first user list carries none.
|
||||
AsksTheVault []string
|
||||
|
||||
// State is the local names of the state this principal's module keeps, and Reads the state of
|
||||
// others it reads as `<module>.<name>` (novox/hq ADR 0201): a bucket each, kept by the owner's
|
||||
// instances and read by whoever declares it.
|
||||
@@ -269,6 +275,14 @@ var VerbsTheControllerAsksTheDeliveryOwner = []SeatVerb{{Seat: "mesh-delivery",
|
||||
{Seat: "mesh-delivery", Verb: "close"}, {Seat: "mesh-delivery", Verb: "release"}, {Seat: "mesh-delivery", Verb: "stop"},
|
||||
{Seat: "mesh-delivery", Verb: "times"}, {Seat: "mesh-delivery", Verb: "groups"}}
|
||||
|
||||
// VerbsTheControllerAsksTheVault are the mesh-vault seat's verbs the controller calls in its own name (novox/hq
|
||||
// ADR 0349 clauses 8 to 10, design 57's V2, issue 675): it carries a value given at its terminal, typed at a desk
|
||||
// or issued by a provider's software into the vault, sealed to a key the vault made for the call, and asks a
|
||||
// rotation. The vault takes these from the controller alone, which it knows from the subject that names its
|
||||
// caller; granted as Principal.AsksTheVault says.
|
||||
var VerbsTheControllerAsksTheVault = []SeatVerb{{Seat: "mesh-vault", Verb: "accept"},
|
||||
{Seat: "mesh-vault", Verb: "secret-ask"}, {Seat: "mesh-vault", Verb: "issued"}, {Seat: "mesh-vault", Verb: "rotate"}}
|
||||
|
||||
// VerbsTheControllerActsOnAWarrant are the other seat verbs the controller calls when the operator's warrant
|
||||
// chooses them (novox/hq ADR 0259): a machine's service restarted, and a walk started or stopped through the
|
||||
// controller's own `plans`. Named one by one; a node seat's on any machine, a mesh seat's flat.
|
||||
@@ -930,6 +944,13 @@ func PermissionsFor(p Principal) (Permissions, error) {
|
||||
}
|
||||
pub = append(pub, "mesh.mod."+module+"."+CallKind+"."+tool+".*."+CallerToken(p.Username()))
|
||||
}
|
||||
// **And the vault's intake verbs** (issue 675): the vault takes a value from the controller alone, known
|
||||
// from the subject's last token — the verbs its holder serves, flat, and nothing else of its seat.
|
||||
for _, v := range VerbsTheControllerAsksTheVault {
|
||||
if slices.Contains(p.AsksTheVault, v.Verb) {
|
||||
pub = append(pub, "mesh.seat."+v.Seat+"."+CallKind+"."+v.Verb+"."+CallerToken(p.Username()))
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
sort.Strings(pub)
|
||||
|
||||
@@ -34,8 +34,6 @@ func TestTheFactsTheGrantPermitsAreTheFactsTheMeshStates(t *testing.T) {
|
||||
states = append(states, link.KeyChecked)
|
||||
// And a walk kept in a new state, for the delivery it walks (novox/hq ADR 0239).
|
||||
states = append(states, link.KeyPlanMoved)
|
||||
// And a key of a record mirror written or removed (novox/hq ADR 0289, design 54 §3; design 57 F3).
|
||||
states = append(states, link.KeyAssignmentChanged, link.KeyDeclaredConfigChanged)
|
||||
for _, event := range states {
|
||||
if !slices.Contains(broker.ControllerStates, event) {
|
||||
t.Errorf("the mesh states %q and its account may not publish it", event)
|
||||
@@ -44,11 +42,6 @@ func TestTheFactsTheGrantPermitsAreTheFactsTheMeshStates(t *testing.T) {
|
||||
if len(broker.ControllerStates) != len(states) {
|
||||
t.Errorf("the grant permits %v, which is more than the mesh states", broker.ControllerStates)
|
||||
}
|
||||
// The record mirrors a module may read are the ones the bus holds (design 54 §5, design 57 F3).
|
||||
if !slices.Equal(catalogue.ControllerMirrors, broker.ControllerMirrors()) {
|
||||
t.Errorf("the catalogue lets a module read the controller's mirrors %v and the bus holds %v",
|
||||
catalogue.ControllerMirrors, broker.ControllerMirrors())
|
||||
}
|
||||
// **And the seat says it.** A seat carries the protocol of its role (novox/hq ADR 0129), so the
|
||||
// facts the control plane states are the seat's `emits` — which is what lets anything else declare
|
||||
// that it consumes them, and what the subject-agreement check reads to know they have an owner.
|
||||
|
||||
@@ -313,10 +313,7 @@ var ControllerStates = []string{"applied", "refused", "built-before",
|
||||
// the pull request's status, and anybody else may read.
|
||||
"checked",
|
||||
// And a walk kept in a new state (novox/hq ADR 0239): what the delivery it walks reads its steps from.
|
||||
"plan-moved",
|
||||
// And a key of a record mirror written or removed (novox/hq ADR 0289, design 54 §3): the assignments and
|
||||
// each registered module's declared config (design 57 F3).
|
||||
"assignment-changed", "declared-config-changed"}
|
||||
"plan-moved"}
|
||||
|
||||
// BusAdvisories are what the bus server says about the mesh's own account that the controller
|
||||
// reads (novox/hq to-be 45 §3, S9): a durable consumer that handed a message over as often as it
|
||||
|
||||
+1
-1
@@ -24,7 +24,7 @@ accounts {
|
||||
jetstream: enabled
|
||||
users = [
|
||||
{ user: "controller", password: "$2a$11$cccccccccccccccccccccc", permissions: {
|
||||
publish: { allow: ["$JS.ACK.CONTROL.controller.>", "$JS.ACK.DEAD_LETTER_NOTICES.controller.>", "$JS.ACK.EVENTS.controller.>", "$JS.API.>", "$KV.SEAT_MESH_BUILD_MACHINE_cancelled.>", "$KV.SEAT_NODE_BUILD_AGENT_cancelled.>", "$KV.mesh-controller_asked.>", "$KV.mesh-controller_assignments.>", "$KV.mesh-controller_calls.>", "$KV.mesh-controller_condition-history.>", "$KV.mesh-controller_conditions.>", "$KV.mesh-controller_declared-config.>", "$KV.mesh-controller_hand-acts.>", "$KV.mesh-controller_lease.>", "$SRV.INFO", "_INBOX.enrol.>", "mesh.again.>", "mesh.assignment.>", "mesh.events.dead.>", "mesh.mod.*.tool.>", "mesh.node.>", "mesh.seat.mesh-build-machine.accept.>", "mesh.seat.mesh-build-machine.tool.>", "mesh.seat.mesh-controller.event.applied", "mesh.seat.mesh-controller.event.assignment-changed", "mesh.seat.mesh-controller.event.built-before", "mesh.seat.mesh-controller.event.checked", "mesh.seat.mesh-controller.event.condition-changed", "mesh.seat.mesh-controller.event.condition-cleared", "mesh.seat.mesh-controller.event.condition-raised", "mesh.seat.mesh-controller.event.declared-config-changed", "mesh.seat.mesh-controller.event.doctor-heartbeat", "mesh.seat.mesh-controller.event.healer-acted", "mesh.seat.mesh-controller.event.plan-moved", "mesh.seat.mesh-controller.event.refused", "mesh.seat.mesh-controller.event.rolled-back", "mesh.seat.mesh-controller.event.secret-replaced", "mesh.seat.mesh-controller.tool.plans", "mesh.seat.mesh-delivery.tool.close", "mesh.seat.mesh-delivery.tool.groups", "mesh.seat.mesh-delivery.tool.release", "mesh.seat.mesh-delivery.tool.stalled", "mesh.seat.mesh-delivery.tool.stop", "mesh.seat.mesh-delivery.tool.times", "mesh.seat.node-backup.tool.backed-up.*", "mesh.seat.node-backup.tool.now.*", "mesh.seat.node-build-agent.accept.>", "mesh.seat.node-build-agent.tool.>", "mesh.seat.node-intrusion-prevention.tool.banned.*", "mesh.seat.node-launcher.tool.secret.*", "mesh.seat.node-service-manager.tool.restart.*"] }
|
||||
publish: { allow: ["$JS.ACK.CONTROL.controller.>", "$JS.ACK.DEAD_LETTER_NOTICES.controller.>", "$JS.ACK.EVENTS.controller.>", "$JS.API.>", "$KV.SEAT_MESH_BUILD_MACHINE_cancelled.>", "$KV.SEAT_NODE_BUILD_AGENT_cancelled.>", "$KV.mesh-controller_asked.>", "$KV.mesh-controller_calls.>", "$KV.mesh-controller_condition-history.>", "$KV.mesh-controller_conditions.>", "$KV.mesh-controller_hand-acts.>", "$KV.mesh-controller_lease.>", "$SRV.INFO", "_INBOX.enrol.>", "mesh.again.>", "mesh.assignment.>", "mesh.events.dead.>", "mesh.mod.*.tool.>", "mesh.node.>", "mesh.seat.mesh-build-machine.accept.>", "mesh.seat.mesh-build-machine.tool.>", "mesh.seat.mesh-controller.event.applied", "mesh.seat.mesh-controller.event.built-before", "mesh.seat.mesh-controller.event.checked", "mesh.seat.mesh-controller.event.condition-changed", "mesh.seat.mesh-controller.event.condition-cleared", "mesh.seat.mesh-controller.event.condition-raised", "mesh.seat.mesh-controller.event.doctor-heartbeat", "mesh.seat.mesh-controller.event.healer-acted", "mesh.seat.mesh-controller.event.plan-moved", "mesh.seat.mesh-controller.event.refused", "mesh.seat.mesh-controller.event.rolled-back", "mesh.seat.mesh-controller.event.secret-replaced", "mesh.seat.mesh-controller.tool.plans", "mesh.seat.mesh-delivery.tool.close", "mesh.seat.mesh-delivery.tool.groups", "mesh.seat.mesh-delivery.tool.release", "mesh.seat.mesh-delivery.tool.stalled", "mesh.seat.mesh-delivery.tool.stop", "mesh.seat.mesh-delivery.tool.times", "mesh.seat.node-backup.tool.backed-up.*", "mesh.seat.node-backup.tool.now.*", "mesh.seat.node-build-agent.accept.>", "mesh.seat.node-build-agent.tool.>", "mesh.seat.node-intrusion-prevention.tool.banned.*", "mesh.seat.node-launcher.tool.secret.*", "mesh.seat.node-service-manager.tool.restart.*"] }
|
||||
subscribe: { allow: ["$JS.API.>", "$JS.EVENT.ADVISORY.CONSUMER.DELETED.>", "$JS.EVENT.ADVISORY.CONSUMER.MAX_DELIVERIES.>", "$SRV.INFO", "$SRV.INFO.mesh-controller", "$SRV.INFO.mesh-controller.>", "$SRV.PING", "$SRV.PING.mesh-controller", "$SRV.PING.mesh-controller.>", "$SRV.STATS", "$SRV.STATS.mesh-controller", "$SRV.STATS.mesh-controller.>", "_DELIVER.controller", "_DELIVER.controller.>", "_INBOX.controller.>", "mesh.control.>", "mesh.mod.*.event.provisioner.failing", "mesh.mod.*.event.provisioner.recovered", "mesh.mod.*.event.provisioner.retirement", "mesh.mod.gitea.event.pull.merged", "mesh.mod.gitea.event.pull.updated", "mesh.mod.mesh-catalog.event.catching-up", "mesh.mod.mesh-catalog.event.upgraded", "mesh.seat.mesh-build-machine.event.built", "mesh.seat.mesh-controller.tool.>", "mesh.seat.node-build-agent.event.built", "mesh.seat.operator-channel.event.decided.mesh-controller"] }
|
||||
allow_responses: { max: 1, ttl: "1m" }
|
||||
} }
|
||||
|
||||
@@ -2,6 +2,7 @@ package broker
|
||||
|
||||
import (
|
||||
"fmt"
|
||||
"slices"
|
||||
"sort"
|
||||
)
|
||||
|
||||
@@ -87,7 +88,7 @@ type Records struct {
|
||||
// is a mesh that cannot be told anything, and there is no state of the records in which that is
|
||||
// correct.
|
||||
func Users(r Records) ([]Principal, error) {
|
||||
out := []Principal{{Kind: KindController, Uses: asksSeatOf(r), AsksIssued: issuedOf(r)}}
|
||||
out := []Principal{{Kind: KindController, Uses: asksSeatOf(r), AsksIssued: issuedOf(r), AsksTheVault: vaultVerbsOf(r)}}
|
||||
|
||||
for _, node := range sortedCopy(r.Nodes) {
|
||||
witness := false
|
||||
@@ -242,3 +243,24 @@ func asksSeatOf(r Records) []Seat {
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
// vaultVerbsOf are the vault's intake verbs its holder serves, as the records say (issue 675): what the controller
|
||||
// is granted to ask it in its own name. None while no holder serves them.
|
||||
func vaultVerbsOf(r Records) []string {
|
||||
seen := map[string]bool{}
|
||||
var out []string
|
||||
for _, node := range sortedCopy(r.Nodes) {
|
||||
for _, d := range r.Assigned[node] {
|
||||
for _, s := range d.Holds {
|
||||
for _, v := range VerbsTheControllerAsksTheVault {
|
||||
if s.Name == v.Seat && slices.Contains(s.Serves, v.Verb) && !seen[v.Verb] {
|
||||
seen[v.Verb] = true
|
||||
out = append(out, v.Verb)
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
sort.Strings(out)
|
||||
return out
|
||||
}
|
||||
|
||||
@@ -158,15 +158,6 @@ var WritersTable = []WriterRow{
|
||||
{State: "a message a consumer gave up on", Writer: "controller", KeptIn: "the bus, the stream " + DeadLettersStream,
|
||||
Others: "read, delivered again or dropped through the controller's dead-letters verb",
|
||||
Subjects: []string{deadLetterPrefix + ">", againPrefix + ">"}, Writes: isController},
|
||||
// The record mirrors (novox/hq ADR 0289 §9, design 54 §2): rows of this table, so a grant that lets
|
||||
// anything else write them is refused at composition. Built so far: the assignments and each registered
|
||||
// module's declared config (design 57 F3).
|
||||
{State: "the assignments mirror", Writer: "controller (lease holder), from its store, only on a change",
|
||||
KeptIn: "key-value " + AssignmentsBucket, Others: "read through `reads` as mesh-controller." + MirrorAssignments,
|
||||
Subjects: kvOf(AssignmentsBucket), Writes: isController},
|
||||
{State: "the declared-config mirror", Writer: "controller (lease holder), from its store, only on a change",
|
||||
KeptIn: "key-value " + DeclaredConfigBucket, Others: "read through `reads` as mesh-controller." + MirrorDeclaredConfig,
|
||||
Subjects: kvOf(DeclaredConfigBucket), Writes: isController},
|
||||
}
|
||||
|
||||
// CheckWriters refuses a grant that lets a principal publish on a subject the writers table gives
|
||||
|
||||
@@ -34,9 +34,6 @@ var designRows = []string{
|
||||
"the operator-channel's open messages",
|
||||
"the facts snapshot",
|
||||
"a message a consumer gave up on",
|
||||
// The record mirrors, rows of to-be 45 §1's table by design 54 §2 (ADR 0289 §9).
|
||||
"the assignments mirror",
|
||||
"the declared-config mirror",
|
||||
}
|
||||
|
||||
func TestTheWritersTableIsTheDesigns(t *testing.T) {
|
||||
|
||||
@@ -107,9 +107,7 @@ var defaultSeats = append([]Seat{
|
||||
// A pull request's merge check, judged (novox/hq to-be 45 §9).
|
||||
"checked",
|
||||
// A walk kept in a new state, for the delivery it walks (novox/hq ADR 0239).
|
||||
"plan-moved",
|
||||
// A key of a record mirror written or removed (novox/hq ADR 0289, design 54 §3; design 57 F3).
|
||||
"assignment-changed", "declared-config-changed"},
|
||||
"plan-moved"},
|
||||
Serves: ControllerVerbs},
|
||||
// The store's first verbs (novox/hq ADR 0159): the smallest set that makes the store askable,
|
||||
// served by whichever module holds the seat with tools of these names.
|
||||
@@ -131,9 +129,11 @@ var defaultSeats = append([]Seat{
|
||||
// test for a seat of the mesh's own (novox/hq ADR 0161) — a second provider of `secret` is a
|
||||
// second claimant, refused by name, rather than a candidate for a pin.
|
||||
//
|
||||
// Its one verb, `get`, hands a secret on request and reply, sealed to the key the asker sends and only
|
||||
// to an asker the vault allows (novox/hq ADR 0349 §11, issue 674). Optional until the vault that runs
|
||||
// serves it (design 33 §7).
|
||||
// Its verb `get` hands a secret on request and reply, sealed to the key the asker sends and only to an
|
||||
// asker the vault allows (novox/hq ADR 0349 §11, issue 674). `accept`, `secret-ask`, `issued` and `rotate`
|
||||
// take a value in and make one anew (ADR 0349 clauses 8 to 10, issue 675): the first three in two steps from
|
||||
// the controller alone, which carries the value sealed to a key the vault made for the call and cannot open
|
||||
// it. All optional until the vault that runs serves them (design 33 §7).
|
||||
{Name: "mesh-vault", Scope: ScopeMesh, Delivers: "secret", Decision: "novox/hq ADR 0161",
|
||||
Serves: []Verb{
|
||||
{Name: "get", Optional: true, Description: "One secret, sealed to the public key the asker sends (the " +
|
||||
@@ -142,6 +142,30 @@ var defaultSeats = append([]Seat{
|
||||
Input: schema(map[string]string{"name": "the secret's name",
|
||||
"public_key": "the asker's X25519 public key as SPKI PEM — never a private key"},
|
||||
[]string{"name", "public_key"})},
|
||||
{Name: "accept", Optional: true, Description: "Take a value given at the controller's terminal, from the " +
|
||||
"controller alone: name the secret and get a call and a key made for it, then send the call and the " +
|
||||
"value sealed to that key. The vault alone opens it, and never remakes an accepted credential.",
|
||||
Input: schema(vaultIntake(map[string]string{"node": "the machine, for a module's own secret",
|
||||
"module": "the module, for a module's own secret", "name": "the secret's name in the module",
|
||||
"provider": "the provider's machine, for a provision's credential", "provision": "the provision",
|
||||
"by": "who gave it at the terminal"}), nil)},
|
||||
{Name: "secret-ask", Optional: true, Description: "Ask the operator for a module's own secret at a desk, " +
|
||||
"from the controller alone: the ask is recorded in the vault's store and bounded before the prompt " +
|
||||
"opens, and what is typed is sealed to a key the vault made for the call and opens there alone.",
|
||||
Input: schema(vaultIntake(map[string]string{"node": "the machine the secret is for",
|
||||
"module": "the module", "name": "the secret's name in the module",
|
||||
"desk": "the machine whose prompt asks", "asked_by": "who asked, as the bus named its caller",
|
||||
"outcome": "instead of sealed: how the prompt ended"}), nil)},
|
||||
{Name: "issued", Optional: true, Description: "Take the credential a provider's software issued for a " +
|
||||
"provision, from the controller alone, which asks the provider's tool with a key the vault made for " +
|
||||
"the call; kept as the provision's one credential, and the same value again writes nothing.",
|
||||
Input: schema(vaultIntake(map[string]string{"provision": "the provision",
|
||||
"provider": "the provider's machine", "module": "the provider's module"}), nil)},
|
||||
{Name: "rotate", Optional: true, Description: "Make a secret anew. Refused for a value a provider's " +
|
||||
"software issued and for an accepted credential. Never answers the value.",
|
||||
Input: schema(map[string]string{"node": "the machine, for a module's own secret", "module": "the module",
|
||||
"name": "the secret's name in the module", "provider": "the provider's machine",
|
||||
"provision": "the provision", "by": "who asked, when the controller carries the act"}, nil)},
|
||||
}},
|
||||
// Named for its scope since 2026-09-30 (novox/hq ADR 0156); `the-artifact-store` resolves to it as
|
||||
// an alias on a mesh that predates the rename. It serves artifacts of every kind a build makes —
|
||||
@@ -824,3 +848,13 @@ func buildAgentVerbs() []Verb {
|
||||
// BrokerSeat is the mesh's bus. Its holder is the bus, and its account — when it declares one — may
|
||||
// snapshot the bus's streams and do nothing else (novox/hq ADR 0235).
|
||||
const BrokerSeat = "mesh-broker"
|
||||
|
||||
// vaultIntake is a two-step vault verb's arguments: what its first step names, then the call and what was sealed.
|
||||
func vaultIntake(first map[string]string) map[string]string {
|
||||
out := map[string]string{"call": "the second step: the call the first step answered",
|
||||
"sealed": "the second step: the value sealed to the call's key — never the value"}
|
||||
for k, v := range first {
|
||||
out[k] = v
|
||||
}
|
||||
return out
|
||||
}
|
||||
|
||||
@@ -124,14 +124,6 @@ func StateProblems(m Manifest) []string {
|
||||
return problems
|
||||
}
|
||||
|
||||
// ControllerMirrors are the controller's record mirrors a module may read, each `mesh-controller.<name>`
|
||||
// (novox/hq ADR 0289, design 54 §5, design 57 F3). **The controller keeps them, not its manifest's `state`**:
|
||||
// they are asserted with the controller's own buckets, with options of their own, and written by the
|
||||
// controller's account alone (the writers table); declared as a module's state, they would be asserted a
|
||||
// second way and granted to the module's own principal as well. Named here so a read of one is a read of
|
||||
// kept state. The bus's list is broker.ControllerMirrors; a test keeps the two agreeing.
|
||||
var ControllerMirrors = []string{"assignments", "declared-config"}
|
||||
|
||||
// StateReadsNothingDeclares is every read across a catalogue whose owner is present and declares no
|
||||
// such state. An absent owner says nothing — a module may be installed long before the one whose
|
||||
// state it reads, as a consumer may before its emitter (design 32 §1).
|
||||
@@ -144,11 +136,6 @@ func StateReadsNothingDeclares(manifests []Manifest) []string {
|
||||
}
|
||||
declared[m.Module] = own
|
||||
}
|
||||
if own, present := declared[ControllerSeatName]; present {
|
||||
for _, name := range ControllerMirrors {
|
||||
own[name] = true
|
||||
}
|
||||
}
|
||||
var problems []string
|
||||
for _, m := range manifests {
|
||||
for _, r := range m.Reads {
|
||||
|
||||
@@ -67,21 +67,6 @@ func TestAReadNamesStateItsOwnerKeeps(t *testing.T) {
|
||||
}
|
||||
}
|
||||
|
||||
// **The controller's record mirrors are kept state** (design 54 §5, design 57 F3): a module may read
|
||||
// `mesh-controller.assignments` and `mesh-controller.declared-config` although the controller's manifest
|
||||
// declares no `state`; a read of anything else of the controller's is still named.
|
||||
func TestAReadOfTheControllersMirrorsIsAReadOfKeptState(t *testing.T) {
|
||||
controller := Manifest{Module: ControllerSeatName}
|
||||
reader := Manifest{Module: "mesh-config", Reads: []string{"mesh-controller.assignments", "mesh-controller.declared-config"}}
|
||||
if p := StateReadsNothingDeclares([]Manifest{controller, reader}); len(p) != 0 {
|
||||
t.Fatalf("a read of the controller's mirrors was refused: %v", p)
|
||||
}
|
||||
other := Manifest{Module: "other", Reads: []string{"mesh-controller.calls"}}
|
||||
if p := StateReadsNothingDeclares([]Manifest{controller, other}); len(p) != 1 {
|
||||
t.Fatalf("a read of the controller's own calls was not named: %v", p)
|
||||
}
|
||||
}
|
||||
|
||||
// **Across the whole catalogue**: every state name is local, and every read whose owner is present
|
||||
// names state that owner keeps.
|
||||
func TestEveryManifestsStateIsLocalAndEveryReadIsKept(t *testing.T) {
|
||||
|
||||
@@ -54,3 +54,46 @@ func TestTheVaultsSeatPromisesGet(t *testing.T) {
|
||||
t.Fatalf("the vault that does not serve get yet was refused: %v", err)
|
||||
}
|
||||
}
|
||||
|
||||
// The vault takes values in and rotates them (novox/hq ADR 0349 clauses 8 to 10, issue 675): the seat promises
|
||||
// accept, secret-ask, issued and rotate, each with the arguments of its steps; a holder serving them is accepted,
|
||||
// and, optional while the vault that runs does not, so is one that does not yet.
|
||||
func TestTheVaultsSeatPromisesItsIntakeVerbs(t *testing.T) {
|
||||
seat, _ := SeatNamed("mesh-vault")
|
||||
want := map[string][]string{
|
||||
"accept": {"node", "module", "name", "provider", "provision", "by", "call", "sealed"},
|
||||
"secret-ask": {"node", "module", "name", "desk", "asked_by", "outcome", "call", "sealed"},
|
||||
"issued": {"provision", "provider", "module", "call", "sealed"},
|
||||
"rotate": {"node", "module", "name", "provider", "provision", "by"},
|
||||
}
|
||||
for name, args := range want {
|
||||
var verb *Verb
|
||||
for i := range seat.Serves {
|
||||
if seat.Serves[i].Name == name {
|
||||
verb = &seat.Serves[i]
|
||||
}
|
||||
}
|
||||
if verb == nil {
|
||||
t.Errorf("the vault's seat does not promise %s", name)
|
||||
continue
|
||||
}
|
||||
if !verb.Optional {
|
||||
t.Errorf("%s is not optional while the vault that runs does not serve it", name)
|
||||
}
|
||||
props, _ := verb.Input["properties"].(map[string]any)
|
||||
for _, arg := range args {
|
||||
if props[arg] == nil {
|
||||
t.Errorf("%s takes no %s", name, arg)
|
||||
}
|
||||
}
|
||||
if props["value"] != nil {
|
||||
t.Errorf("%s takes a value in the clear", name)
|
||||
}
|
||||
}
|
||||
serving := Manifest{Module: "mesh-vault", Provides: []Offer{{Name: "secret", Scope: ScopeMesh}},
|
||||
Claims: []Claim{{Name: "mesh-vault", Scope: ScopeMesh,
|
||||
Serves: []string{"get", "accept", "secret-ask", "issued", "rotate"}}}}
|
||||
if err := CanHold(serving, seat); err != nil {
|
||||
t.Fatalf("the vault serving its verbs was refused: %v", err)
|
||||
}
|
||||
}
|
||||
|
||||
@@ -1,153 +0,0 @@
|
||||
package inventory
|
||||
|
||||
import (
|
||||
"bytes"
|
||||
"context"
|
||||
"encoding/json"
|
||||
"fmt"
|
||||
"time"
|
||||
)
|
||||
|
||||
// The values of the controller's record mirrors, built from the store (novox/hq ADR 0289, design 54 §1,
|
||||
// design 57 package F3).
|
||||
//
|
||||
// **Each value is built from a named list of fields, and nothing else** (design 54 §1): no epoch, no time
|
||||
// of writing and no time of the last look, so any lease holder builds the same bytes from the same record
|
||||
// and a pass over unchanged records writes nothing (ADR 0289 §1, §4). The encoding is fixed: the fields in
|
||||
// their struct's order, a declaration's maps sorted.
|
||||
|
||||
// AssignmentMirrored is one assignment as the assignments mirror holds it: the module, the node, and since
|
||||
// when it is assigned there. Design 54 §1 also lists whether it is pending, the builds sent and running and
|
||||
// the seats and provisions it holds there; those join with issue 419, which builds the rest of design 54.
|
||||
type AssignmentMirrored struct {
|
||||
Node string `json:"node"`
|
||||
Module string `json:"module"`
|
||||
Assigned time.Time `json:"assigned"`
|
||||
}
|
||||
|
||||
// DeclaredConfigMirrored is one registered module's config declarations as the declared-config mirror holds
|
||||
// them: the module, and each key its manifest declares — under `config`, or under its old name `settings` for a
|
||||
// module not yet moved (ADR 0348 §1) — with that declaration's listed fields. An empty object when the module
|
||||
// declares none: registered, declaring nothing.
|
||||
type DeclaredConfigMirrored struct {
|
||||
Module string `json:"module"`
|
||||
Settings map[string]DeclaredMirrored `json:"settings"`
|
||||
}
|
||||
|
||||
// DeclaredMirrored is one key's declaration as the mirror holds it: **its listed fields and no other**
|
||||
// (ADR 0289 §12), and every field a declaration has (catalogue.SettingDeclaration), under the same names. The
|
||||
// mirror holds what the module's manifest declares about its config values, and mesh-config validates and
|
||||
// resolves from it (ADR 0356 §4, ADR 0348 §4, §9), so it carries the whole declaration: kind, default and why
|
||||
// of the settings form, and type, description, validation, trusted, predefined, required and runtime of the
|
||||
// config form. None is a secret: a secret lives in the vault, never in config, and the catalogue refuses a
|
||||
// default for a key whose name says secret. A field the manifest gains reaches the bus only once it is added
|
||||
// here, deliberately: a test fails until it is listed. The config form's fields are left out when unset, so a
|
||||
// module not yet moved is mirrored as it was before they were carried.
|
||||
type DeclaredMirrored struct {
|
||||
Kind string `json:"kind"`
|
||||
Default json.RawMessage `json:"default"`
|
||||
Why string `json:"why"`
|
||||
Type string `json:"type,omitempty"`
|
||||
Description string `json:"description,omitempty"`
|
||||
Enum json.RawMessage `json:"enum,omitempty"`
|
||||
Minimum *float64 `json:"minimum,omitempty"`
|
||||
Maximum *float64 `json:"maximum,omitempty"`
|
||||
Pattern string `json:"pattern,omitempty"`
|
||||
Trusted bool `json:"trusted,omitempty"`
|
||||
Predefined string `json:"predefined,omitempty"`
|
||||
Required bool `json:"required,omitempty"`
|
||||
Runtime bool `json:"runtime,omitempty"`
|
||||
}
|
||||
|
||||
// DeclaredMirroredFields are the fields of a declaration the mirror carries, by their manifest names.
|
||||
var DeclaredMirroredFields = []string{"kind", "default", "why", "type", "description", "enum", "minimum", "maximum",
|
||||
"pattern", "trusted", "predefined", "required", "runtime"}
|
||||
|
||||
// AssignmentsMirror is every assignment's mirrored value, keyed `<node>.<module>`.
|
||||
func (i *Inventory) AssignmentsMirror(ctx context.Context) (map[string][]byte, error) {
|
||||
rows, err := i.store.Pool().Query(ctx,
|
||||
`select n.name, a.module, a.assigned from assignment a join node n on n.id = a.node`)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
defer rows.Close()
|
||||
out := map[string][]byte{}
|
||||
for rows.Next() {
|
||||
var a AssignmentMirrored
|
||||
if err := rows.Scan(&a.Node, &a.Module, &a.Assigned); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
a.Assigned = a.Assigned.UTC()
|
||||
value, err := json.Marshal(a)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
out[a.Node+"."+a.Module] = value
|
||||
}
|
||||
return out, rows.Err()
|
||||
}
|
||||
|
||||
// DeclaredConfigMirror is every registered module's mirrored config declarations, keyed by the module.
|
||||
func (i *Inventory) DeclaredConfigMirror(ctx context.Context) (map[string][]byte, error) {
|
||||
rows, err := i.store.Pool().Query(ctx,
|
||||
`select name, coalesce(manifest->'config', manifest->'settings', '{}'::jsonb) from module`)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
defer rows.Close()
|
||||
out := map[string][]byte{}
|
||||
for rows.Next() {
|
||||
var (
|
||||
d DeclaredConfigMirrored
|
||||
raw []byte
|
||||
)
|
||||
if err := rows.Scan(&d.Module, &raw); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
if d.Settings, err = declaredMirrored(raw); err != nil {
|
||||
return nil, fmt.Errorf("the declarations of %s cannot be read: %w", d.Module, err)
|
||||
}
|
||||
value, err := json.Marshal(d)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
out[d.Module] = value
|
||||
}
|
||||
return out, rows.Err()
|
||||
}
|
||||
|
||||
// declaredMirrored is a manifest's config declarations, keyed by the value, as the mirror holds them: each
|
||||
// with its listed fields, a default and an enum encoded the one way.
|
||||
func declaredMirrored(raw []byte) (map[string]DeclaredMirrored, error) {
|
||||
var declared map[string]DeclaredMirrored
|
||||
if err := json.Unmarshal(raw, &declared); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
out := map[string]DeclaredMirrored{}
|
||||
for key, decl := range declared {
|
||||
var err error
|
||||
if len(decl.Default) > 0 {
|
||||
if decl.Default, err = canonical(decl.Default); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
}
|
||||
if len(decl.Enum) > 0 {
|
||||
if decl.Enum, err = canonical(decl.Enum); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
}
|
||||
out[key] = decl
|
||||
}
|
||||
return out, nil
|
||||
}
|
||||
|
||||
// canonical is a JSON document encoded the one way: objects' keys sorted, numbers as written.
|
||||
func canonical(raw []byte) ([]byte, error) {
|
||||
dec := json.NewDecoder(bytes.NewReader(raw))
|
||||
dec.UseNumber()
|
||||
var v any
|
||||
if err := dec.Decode(&v); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
return json.Marshal(v)
|
||||
}
|
||||
@@ -1,113 +0,0 @@
|
||||
package inventory
|
||||
|
||||
import (
|
||||
"encoding/json"
|
||||
"reflect"
|
||||
"slices"
|
||||
"strings"
|
||||
"testing"
|
||||
|
||||
"github.com/novox/mesh-controller/internal/catalogue"
|
||||
)
|
||||
|
||||
// jsonFields is the names a struct's fields are encoded under.
|
||||
func jsonFields(t reflect.Type) []string {
|
||||
var out []string
|
||||
for i := 0; i < t.NumField(); i++ {
|
||||
name, _, _ := strings.Cut(t.Field(i).Tag.Get("json"), ",")
|
||||
if name != "" && name != "-" {
|
||||
out = append(out, name)
|
||||
}
|
||||
}
|
||||
return out
|
||||
}
|
||||
|
||||
// **A declaration reaches the bus with its listed fields and no other** (ADR 0289 §12): the mirror's value
|
||||
// carries exactly DeclaredMirroredFields, and every field a manifest's declaration has is on that list. A
|
||||
// field a declaration gains (package K1 widens it, ADR 0348 clause 4) fails here until it is listed — read,
|
||||
// judged not a secret and carried — so nothing reaches the declared-config mirror unreviewed.
|
||||
func TestTheDeclaredConfigMirrorCarriesOnlyItsListedFields(t *testing.T) {
|
||||
if got := jsonFields(reflect.TypeOf(DeclaredMirrored{})); !slices.Equal(got, DeclaredMirroredFields) {
|
||||
t.Errorf("the mirrored declaration carries %v, its list is %v", got, DeclaredMirroredFields)
|
||||
}
|
||||
for _, field := range jsonFields(reflect.TypeOf(catalogue.SettingDeclaration{})) {
|
||||
if !slices.Contains(DeclaredMirroredFields, field) {
|
||||
t.Errorf("a declaration has the field %q, which the declared-config mirror does not list: list it "+
|
||||
"in DeclaredMirrored (and say why it is no secret), or say here why it stays off the bus", field)
|
||||
}
|
||||
}
|
||||
if got := jsonFields(reflect.TypeOf(DeclaredConfigMirrored{})); !slices.Equal(got, []string{"module", "settings"}) {
|
||||
t.Errorf("the mirrored module carries %v", got)
|
||||
}
|
||||
}
|
||||
|
||||
// **A declaration using every field reaches the mirror whole** (ADR 0356 §4): the mirror holds what the manifest
|
||||
// declares about a config value, so a declaration with every field set reads back from the mirror's value with
|
||||
// each field as written.
|
||||
func TestADeclarationUsingEveryFieldRoundTripsThroughTheMirror(t *testing.T) {
|
||||
low, high := 1.0, 65535.0
|
||||
every := catalogue.SettingDeclaration{Kind: catalogue.KindPreference, Default: 8080, Why: "the port it always had",
|
||||
Type: catalogue.ConfigNumber, Description: "the port it listens on, a TCP port number",
|
||||
Enum: []any{8080, 8443}, Minimum: &low, Maximum: &high, Pattern: "^[0-9]+$", Trusted: true,
|
||||
Predefined: "http.port", Required: true, Runtime: true}
|
||||
declared, err := json.Marshal(map[string]catalogue.SettingDeclaration{"port": every})
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
var written map[string]map[string]any
|
||||
if err := json.Unmarshal(declared, &written); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
for _, field := range jsonFields(reflect.TypeOf(catalogue.SettingDeclaration{})) {
|
||||
if _, has := written["port"][field]; !has {
|
||||
t.Fatalf("the declaration under test leaves %q unset: set every field", field)
|
||||
}
|
||||
}
|
||||
mirrored, err := declaredMirrored(declared)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
value, err := json.Marshal(DeclaredConfigMirrored{Module: "proxy", Settings: mirrored})
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
var read struct {
|
||||
Settings map[string]map[string]any `json:"settings"`
|
||||
}
|
||||
if err := json.Unmarshal(value, &read); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if !reflect.DeepEqual(read.Settings, written) {
|
||||
t.Errorf("the mirror holds %v; the manifest declared %v", read.Settings, written)
|
||||
}
|
||||
}
|
||||
|
||||
// **A module moved to the config form is mirrored from it** (ADR 0348 §1): the store keeps its manifest in the
|
||||
// form it was written in, under `config`, and the declared-config mirror reads its declarations from there with
|
||||
// the config form's fields.
|
||||
func TestAModuleInTheConfigFormIsMirroredWithItsWholeDeclaration(t *testing.T) {
|
||||
inv := fresh(t)
|
||||
var m catalogue.Manifest
|
||||
if err := json.Unmarshal([]byte(`{"module": "proxy", "version": "1", "config": {"port": {"type": "number",
|
||||
"description": "the port it listens on, a TCP port number", "default": 8080, "minimum": 1,
|
||||
"maximum": 65535, "enum": [8080, 8443], "predefined": "http.port"}}}`), &m); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if err := inv.RegisterModule(t.Context(), m, Source{}); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
values, err := inv.DeclaredConfigMirror(t.Context())
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
var got DeclaredConfigMirrored
|
||||
if err := json.Unmarshal(values["proxy"], &got); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
port, has := got.Settings["port"]
|
||||
if !has || port.Type != catalogue.ConfigNumber || port.Description == "" || string(port.Default) != "8080" ||
|
||||
port.Minimum == nil || *port.Minimum != 1 || port.Maximum == nil || *port.Maximum != 65535 ||
|
||||
string(port.Enum) != "[8080,8443]" || port.Predefined != "http.port" {
|
||||
t.Errorf("a module in the config form is mirrored as %s", values["proxy"])
|
||||
}
|
||||
}
|
||||
@@ -81,11 +81,6 @@ const (
|
||||
// steps from. Said after every save of the serving controller; a delivery's holder also asks for the
|
||||
// walks it follows, so a save made elsewhere is found by comparison, never lost.
|
||||
KeyPlanMoved = "plan-moved"
|
||||
// KeyAssignmentChanged and KeyDeclaredConfigChanged: a key of the controller's assignments or declared-config
|
||||
// mirror was written or removed (novox/hq ADR 0289, design 54 §3, design 57 F3): which key, how and at which
|
||||
// revision of the bucket — never the value, which the bucket holds.
|
||||
KeyAssignmentChanged = "assignment-changed"
|
||||
KeyDeclaredConfigChanged = "declared-config-changed"
|
||||
)
|
||||
|
||||
// Applied is what a machine now runs, as the mesh states it.
|
||||
|
||||
+17
-1
@@ -609,13 +609,29 @@ func PausedSaid(js *broker.JetStream, seat string, nodes []string) (map[string]H
|
||||
// (design 33 §4: a mesh seat's verb carries no machine), and reads its answer (novox/hq ADR 0239: the
|
||||
// delivery's owner's `stalled` and `close`).
|
||||
func AskMeshSeatTool(ctx context.Context, conn *nats.Conn, seat, verb string, args any, timeout time.Duration) (Answer, error) {
|
||||
return askMeshSeatToolAt(ctx, conn, SeatToolSubject(seat, verb), seat, verb, args, timeout)
|
||||
}
|
||||
|
||||
// AskMeshSeatToolAs is AskMeshSeatTool in the caller's own name: on the subject that names its caller (novox/hq
|
||||
// issue 365), which the bus lets only that user publish, so the holder knows who asks from the subject alone —
|
||||
// how the vault takes a value from the controller and nobody else (design 57's V2, issue 675).
|
||||
func AskMeshSeatToolAs(ctx context.Context, conn *nats.Conn, caller, seat, verb string, args any,
|
||||
timeout time.Duration) (Answer, error) {
|
||||
subject := broker.CalledSubject(SeatToolSubject(seat, verb), caller)
|
||||
if subject == "" {
|
||||
return Answer{}, fmt.Errorf("%q cannot ask %s.%s in its own name: it is not a bus user's name", caller, seat, verb)
|
||||
}
|
||||
return askMeshSeatToolAt(ctx, conn, subject, seat, verb, args, timeout)
|
||||
}
|
||||
|
||||
func askMeshSeatToolAt(ctx context.Context, conn *nats.Conn, subject, seat, verb string, args any,
|
||||
timeout time.Duration) (Answer, error) {
|
||||
body, err := json.Marshal(args)
|
||||
if err != nil {
|
||||
return Answer{}, err
|
||||
}
|
||||
asking, cancel := context.WithTimeout(ctx, timeout)
|
||||
defer cancel()
|
||||
subject := SeatToolSubject(seat, verb)
|
||||
refused, stop := refusalsOf(conn, subject)
|
||||
defer stop()
|
||||
type replied struct {
|
||||
|
||||
@@ -0,0 +1,47 @@
|
||||
package link
|
||||
|
||||
import (
|
||||
"context"
|
||||
"encoding/json"
|
||||
"strings"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
"github.com/nats-io/nats.go"
|
||||
|
||||
"github.com/novox/mesh-controller/internal/broker"
|
||||
"github.com/novox/mesh-controller/internal/testbus"
|
||||
)
|
||||
|
||||
// A mesh seat's verb asked in the caller's own name goes on the subject that names its caller (novox/hq issue
|
||||
// 365), so the holder knows who asks from the subject alone: how the vault takes a value from the controller
|
||||
// and nobody else (design 57's V2, issue 675).
|
||||
func TestAMeshSeatsVerbIsAskedInTheCallersOwnName(t *testing.T) {
|
||||
conn, err := nats.Connect(testbus.URL(t))
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
defer conn.Close()
|
||||
heard := make(chan string, 1)
|
||||
sub, err := conn.Subscribe(broker.CalledPattern(SeatToolSubject("mesh-vault", "accept")), func(m *nats.Msg) {
|
||||
heard <- m.Subject
|
||||
raw, _ := json.Marshal(Answer{Result: json.RawMessage(`{"call":"c1"}`)})
|
||||
_ = m.Respond(raw)
|
||||
})
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
defer sub.Unsubscribe()
|
||||
answer, err := AskMeshSeatToolAs(context.Background(), conn, broker.ControllerName, "mesh-vault", "accept",
|
||||
map[string]any{"provider": "novox", "provision": "postgres"}, 5*time.Second)
|
||||
if err != nil || string(answer.Result) != `{"call":"c1"}` {
|
||||
t.Fatalf("%+v %v", answer, err)
|
||||
}
|
||||
subject := <-heard
|
||||
if subject != "mesh.seat.mesh-vault.call.accept.controller" || !strings.HasSuffix(subject, "."+broker.ControllerName) {
|
||||
t.Errorf("asked on %s", subject)
|
||||
}
|
||||
if _, err := AskMeshSeatToolAs(context.Background(), conn, "not a user!", "mesh-vault", "accept", nil, time.Second); err == nil {
|
||||
t.Error("a caller that is no bus user was asked in")
|
||||
}
|
||||
}
|
||||
@@ -1,344 +0,0 @@
|
||||
// Package recordmirror writes the controller's record mirrors (novox/hq ADR 0289, design 54): a bucket per kind
|
||||
// of record, one key per record, written from the store only when the record changed, and one event on the
|
||||
// controller's seat per write.
|
||||
//
|
||||
// **The store stays the record.** A mirror is built from what the store committed, so a change rolled back
|
||||
// is never written. Every pass reads what the bucket holds and compares it byte for byte with what the store
|
||||
// says: a pass over unchanged records writes nothing and says nothing (ADR 0289 §1), and a value somebody
|
||||
// else wrote behind the writer's back (a former lease holder) differs from the store and is put right on
|
||||
// the next pass. Each write names the revision it read (compare-and-set), so a write in between is noticed,
|
||||
// never overwritten blind.
|
||||
//
|
||||
// **Only the lease holder writes** (ADR 0289 §9, design 54 §2): a pass begins by asking whether this
|
||||
// controller holds the lease, and asks again before every write; without it, it writes nothing.
|
||||
//
|
||||
// **The bucket is the authority, the event a notice** (design 54 §3): the event is sent after the write
|
||||
// succeeded, so there is no event for a value the bucket does not hold. A refused event is tried once more
|
||||
// and then counted as lost; a watcher that must not miss a change watches the bucket.
|
||||
//
|
||||
// What this package does not do yet is design 54 §4's audit (probe D17 and its healer H6), which compares a
|
||||
// mirror with its store behind the writer's back; novox/hq issue 419 carries it.
|
||||
package recordmirror
|
||||
|
||||
import (
|
||||
"bytes"
|
||||
"context"
|
||||
"encoding/json"
|
||||
"errors"
|
||||
"fmt"
|
||||
"regexp"
|
||||
"sort"
|
||||
"time"
|
||||
|
||||
"github.com/nats-io/nats.go/jetstream"
|
||||
|
||||
"github.com/novox/mesh-controller/internal/broker"
|
||||
"github.com/novox/mesh-controller/internal/link"
|
||||
)
|
||||
|
||||
// A Mirror is one kind of record the controller keeps on the bus: its local name, its bucket and the event
|
||||
// each write raises on the controller's seat.
|
||||
type Mirror struct {
|
||||
Name string
|
||||
Bucket string
|
||||
Event string
|
||||
}
|
||||
|
||||
// The mirrors this package writes (novox/hq design 57, package F3). The assignments mirror is design 54's,
|
||||
// keyed `<node>.<module>`; the declared config is each registered module's config declarations as its
|
||||
// manifest holds them, keyed by the module, which mesh-config resolves a module's config values from
|
||||
// (ADR 0348 clauses 2 and 4).
|
||||
var (
|
||||
Assignments = Mirror{Name: broker.MirrorAssignments, Bucket: broker.AssignmentsBucket,
|
||||
Event: link.KeyAssignmentChanged}
|
||||
DeclaredConfig = Mirror{Name: broker.MirrorDeclaredConfig, Bucket: broker.DeclaredConfigBucket,
|
||||
Event: link.KeyDeclaredConfigChanged}
|
||||
)
|
||||
|
||||
// Mirrors is every mirror this package writes, in the order they are written.
|
||||
func Mirrors() []Mirror { return []Mirror{Assignments, DeclaredConfig} }
|
||||
|
||||
// The changes an event says.
|
||||
const (
|
||||
Added = "added"
|
||||
Changed = "changed"
|
||||
Removed = "removed"
|
||||
)
|
||||
|
||||
// Event is what one write says on the controller's seat (design 54 §3): which mirror and key, what kind of
|
||||
// change, the names of the top-level fields that changed, the bucket's revision of the write, the lease's
|
||||
// epoch and when. Never the value: a watcher reads it from the bucket at that revision or later.
|
||||
type Event struct {
|
||||
Mirror string `json:"mirror"`
|
||||
Key string `json:"key"`
|
||||
Change string `json:"change"`
|
||||
Fields []string `json:"fields,omitempty"`
|
||||
Revision uint64 `json:"revision"`
|
||||
Epoch uint64 `json:"epoch,omitempty"`
|
||||
At time.Time `json:"at"`
|
||||
}
|
||||
|
||||
// Announcer says an event on a seat; link.OverNATS is one, and so is link.Bus.
|
||||
type Announcer interface {
|
||||
PublishSeatEvent(ctx context.Context, seat, event string, body []byte) error
|
||||
}
|
||||
|
||||
// MarkerTTL is how long the marker a purge leaves is kept (design 54 §2): a day, so the markers of removed
|
||||
// records do not pile up in a bucket that has no age.
|
||||
const MarkerTTL = 24 * time.Hour
|
||||
|
||||
// ErrNotHeld is a pass refused because this controller does not hold the lease: it wrote nothing.
|
||||
var ErrNotHeld = errors.New("this controller does not hold the lease, so it writes no mirror")
|
||||
|
||||
// Writer writes one mirror. Not safe for use by two goroutines at once: one controller's one loop owns it.
|
||||
type Writer struct {
|
||||
Mirror Mirror
|
||||
KV jetstream.KeyValue
|
||||
Announce Announcer
|
||||
// Epoch is the lease's epoch the events carry; nil carries none.
|
||||
Epoch func() uint64
|
||||
// Holds says this controller holds the lease. Asked before every pass and every write; a writer
|
||||
// without it writes nothing (design 54 §2). Required: nil holds nothing.
|
||||
Holds func() bool
|
||||
// Now is the clock the events are dated by; nil is time.Now.
|
||||
Now func() time.Time
|
||||
|
||||
known map[string]kept
|
||||
// LostEvents counts the events refused twice since the writer was made.
|
||||
LostEvents int
|
||||
}
|
||||
|
||||
type kept struct {
|
||||
value []byte
|
||||
revision uint64
|
||||
}
|
||||
|
||||
// Result is what one pass did: keys written, keys removed, and keys refused with why.
|
||||
type Result struct {
|
||||
Written int
|
||||
Removed int
|
||||
Refused int
|
||||
}
|
||||
|
||||
// Sync makes the bucket hold want, one key per record: writes each value that differs from what the key
|
||||
// holds, purges each key no record is left for, and raises one event per write. A key the bus cannot hold
|
||||
// is refused alone, and the others are written; the error then names it.
|
||||
func (w *Writer) Sync(ctx context.Context, want map[string][]byte) (Result, error) {
|
||||
var r Result
|
||||
if !w.holds() {
|
||||
return r, ErrNotHeld
|
||||
}
|
||||
// What the bucket holds now, every pass: a value written behind this writer's back differs from the
|
||||
// store's and is written again below.
|
||||
if err := w.fill(ctx); err != nil {
|
||||
return r, fmt.Errorf("reading the %s mirror: %w", w.Mirror.Name, err)
|
||||
}
|
||||
var problems []error
|
||||
keys := make([]string, 0, len(want))
|
||||
for k := range want {
|
||||
keys = append(keys, k)
|
||||
}
|
||||
sort.Strings(keys)
|
||||
for _, key := range keys {
|
||||
value := want[key]
|
||||
before, had := w.known[key]
|
||||
if had && bytes.Equal(before.value, value) {
|
||||
continue
|
||||
}
|
||||
if err := validKey(key); err != nil {
|
||||
r.Refused++
|
||||
problems = append(problems, err)
|
||||
continue
|
||||
}
|
||||
if !w.holds() {
|
||||
return r, errors.Join(append(problems, ErrNotHeld)...)
|
||||
}
|
||||
revision, wrote, err := w.put(ctx, key, value, before, had)
|
||||
if err != nil {
|
||||
problems = append(problems, fmt.Errorf("writing %s: %w", key, err))
|
||||
continue
|
||||
}
|
||||
w.known[key] = kept{value: append([]byte(nil), value...), revision: revision}
|
||||
if !wrote {
|
||||
continue
|
||||
}
|
||||
r.Written++
|
||||
change, fields := Added, []string(nil)
|
||||
if had {
|
||||
change, fields = Changed, changedFields(before.value, value)
|
||||
}
|
||||
w.say(ctx, key, change, fields, revision)
|
||||
}
|
||||
gone := make([]string, 0)
|
||||
for key := range w.known {
|
||||
if _, ok := want[key]; !ok {
|
||||
gone = append(gone, key)
|
||||
}
|
||||
}
|
||||
sort.Strings(gone)
|
||||
for _, key := range gone {
|
||||
k := w.known[key]
|
||||
if !w.holds() {
|
||||
return r, errors.Join(append(problems, ErrNotHeld)...)
|
||||
}
|
||||
err := w.KV.Purge(ctx, key, jetstream.LastRevision(k.revision), jetstream.PurgeTTL(MarkerTTL))
|
||||
if refusedByRevision(err) {
|
||||
// Read it again: purged already, or moved by another writer.
|
||||
if e, gerr := w.KV.Get(ctx, key); errors.Is(gerr, jetstream.ErrKeyNotFound) {
|
||||
delete(w.known, key)
|
||||
continue
|
||||
} else if gerr == nil && w.holds() {
|
||||
err = w.KV.Purge(ctx, key, jetstream.LastRevision(e.Revision()), jetstream.PurgeTTL(MarkerTTL))
|
||||
}
|
||||
}
|
||||
if err != nil {
|
||||
problems = append(problems, fmt.Errorf("removing %s: %w", key, err))
|
||||
continue
|
||||
}
|
||||
delete(w.known, key)
|
||||
r.Removed++
|
||||
revision := uint64(0)
|
||||
if status, err := w.KV.Status(ctx); err == nil {
|
||||
// The purge's own sequence is the stream's last one only when nothing wrote in between, which no
|
||||
// other writer may; said as the bucket's revision after it.
|
||||
if s, ok := status.(interface{ StreamInfo() *jetstream.StreamInfo }); ok {
|
||||
revision = s.StreamInfo().State.LastSeq
|
||||
}
|
||||
}
|
||||
w.say(ctx, key, Removed, nil, revision)
|
||||
}
|
||||
return r, errors.Join(problems...)
|
||||
}
|
||||
|
||||
// put writes one value by compare-and-set and answers its revision, and whether this call wrote it (false
|
||||
// when the bucket already held it).
|
||||
func (w *Writer) put(ctx context.Context, key string, value []byte, before kept, had bool) (uint64, bool, error) {
|
||||
var (
|
||||
revision uint64
|
||||
err error
|
||||
)
|
||||
if had {
|
||||
revision, err = w.KV.Update(ctx, key, value, before.revision)
|
||||
} else {
|
||||
revision, err = w.KV.Create(ctx, key, value)
|
||||
}
|
||||
if !refusedByRevision(err) {
|
||||
return revision, err == nil, err
|
||||
}
|
||||
// Refused: another writer in between, or this writer's own write whose acknowledgement was lost.
|
||||
e, gerr := w.KV.Get(ctx, key)
|
||||
switch {
|
||||
case errors.Is(gerr, jetstream.ErrKeyNotFound):
|
||||
if !w.holds() {
|
||||
return 0, false, errors.New("the key moved and this controller no longer holds the lease")
|
||||
}
|
||||
revision, err = w.KV.Create(ctx, key, value)
|
||||
return revision, err == nil, err
|
||||
case gerr != nil:
|
||||
return 0, false, gerr
|
||||
case bytes.Equal(e.Value(), value):
|
||||
return e.Revision(), false, nil
|
||||
case !w.holds():
|
||||
return 0, false, errors.New("the key moved and this controller no longer holds the lease")
|
||||
}
|
||||
revision, err = w.KV.Update(ctx, key, value, e.Revision())
|
||||
return revision, err == nil, err
|
||||
}
|
||||
|
||||
// say raises one write's event, tried once more when refused and then counted as lost.
|
||||
func (w *Writer) say(ctx context.Context, key, change string, fields []string, revision uint64) {
|
||||
if w.Announce == nil {
|
||||
return
|
||||
}
|
||||
e := Event{Mirror: w.Mirror.Name, Key: key, Change: change, Fields: fields, Revision: revision, At: w.now().UTC()}
|
||||
if w.Epoch != nil {
|
||||
e.Epoch = w.Epoch()
|
||||
}
|
||||
body, err := json.Marshal(e)
|
||||
if err != nil {
|
||||
w.LostEvents++
|
||||
return
|
||||
}
|
||||
for try := 0; try < 2; try++ {
|
||||
if err = w.Announce.PublishSeatEvent(ctx, broker.ControllerSeat, w.Mirror.Event, body); err == nil {
|
||||
return
|
||||
}
|
||||
}
|
||||
w.LostEvents++
|
||||
}
|
||||
|
||||
// fill reads what the bucket holds now, so a controller started again writes only what differs.
|
||||
func (w *Writer) fill(ctx context.Context) error {
|
||||
w.known = map[string]kept{}
|
||||
watch, err := w.KV.WatchAll(ctx, jetstream.IgnoreDeletes())
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
defer func() { _ = watch.Stop() }()
|
||||
for {
|
||||
select {
|
||||
case <-ctx.Done():
|
||||
return ctx.Err()
|
||||
case e := <-watch.Updates():
|
||||
if e == nil {
|
||||
return nil
|
||||
}
|
||||
w.known[e.Key()] = kept{value: append([]byte(nil), e.Value()...), revision: e.Revision()}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
func (w *Writer) holds() bool { return w.Holds != nil && w.Holds() }
|
||||
|
||||
func (w *Writer) now() time.Time {
|
||||
if w.Now == nil {
|
||||
return time.Now()
|
||||
}
|
||||
return w.Now()
|
||||
}
|
||||
|
||||
// refusedByRevision is the server refusing a compare-and-set: the key is not at the revision named.
|
||||
func refusedByRevision(err error) bool {
|
||||
if err == nil {
|
||||
return false
|
||||
}
|
||||
if errors.Is(err, jetstream.ErrKeyExists) {
|
||||
return true
|
||||
}
|
||||
var apiErr *jetstream.APIError
|
||||
return errors.As(err, &apiErr) && apiErr.ErrorCode == jetstream.JSErrCodeStreamWrongLastSequence
|
||||
}
|
||||
|
||||
// keyChars is what a bucket's key may be made of.
|
||||
var keyChars = regexp.MustCompile(`^[-/_=.a-zA-Z0-9]+$`)
|
||||
|
||||
// validKey refuses a key the bus cannot hold, naming it.
|
||||
func validKey(key string) error {
|
||||
if key == "" || !keyChars.MatchString(key) || key[0] == '.' || key[len(key)-1] == '.' {
|
||||
return fmt.Errorf("%q cannot be a key of a bucket: letters, digits, '-', '_', '=', '/' and '.', not at "+
|
||||
"either end", key)
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
// changedFields names the top-level fields whose values differ between two encoded objects, sorted; nil
|
||||
// when either is not an object.
|
||||
func changedFields(before, after []byte) []string {
|
||||
var a, b map[string]json.RawMessage
|
||||
if json.Unmarshal(before, &a) != nil || json.Unmarshal(after, &b) != nil || a == nil || b == nil {
|
||||
return nil
|
||||
}
|
||||
var out []string
|
||||
for k, v := range b {
|
||||
if old, ok := a[k]; !ok || !bytes.Equal(old, v) {
|
||||
out = append(out, k)
|
||||
}
|
||||
}
|
||||
for k := range a {
|
||||
if _, ok := b[k]; !ok {
|
||||
out = append(out, k)
|
||||
}
|
||||
}
|
||||
sort.Strings(out)
|
||||
return out
|
||||
}
|
||||
@@ -1,178 +0,0 @@
|
||||
package recordmirror
|
||||
|
||||
import (
|
||||
"encoding/json"
|
||||
"strings"
|
||||
"testing"
|
||||
|
||||
"github.com/novox/mesh-controller/internal/broker"
|
||||
"github.com/novox/mesh-controller/internal/catalogue"
|
||||
"github.com/novox/mesh-controller/internal/inventory"
|
||||
)
|
||||
|
||||
// **Design 57's F3, end to end** (novox/hq issue 652): from the controller's store to a real bus. An
|
||||
// assignment changes exactly one key of the assignments mirror and raises exactly one `assignment-changed`;
|
||||
// a registration changes exactly one key of the declared-config mirror and raises exactly one
|
||||
// `declared-config-changed`; and the same records written again change nothing and say nothing — a module
|
||||
// registered again unchanged included, although the store's registration time moved.
|
||||
func TestNatsAnAssignmentAndARegistrationEachChangeOneKeyAndRaiseOneEvent(t *testing.T) {
|
||||
inv := inventory.ForTest(t)
|
||||
ctx := t.Context()
|
||||
js, api := aMirrorBus(t)
|
||||
assignments := aWriter(t, js, api, Assignments)
|
||||
declared := aWriter(t, js, api, DeclaredConfig)
|
||||
heardAssignments := listen(t, js.Conn(), "mesh.seat."+broker.ControllerSeat+".event."+Assignments.Event)
|
||||
heardDeclared := listen(t, js.Conn(), "mesh.seat."+broker.ControllerSeat+".event."+DeclaredConfig.Event)
|
||||
|
||||
pass := func() (Result, Result) {
|
||||
t.Helper()
|
||||
a, err := inv.AssignmentsMirror(ctx)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
ra, err := assignments.Sync(ctx, a)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
d, err := inv.DeclaredConfigMirror(ctx)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
rd, err := declared.Sync(ctx, d)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
return ra, rd
|
||||
}
|
||||
|
||||
if _, err := inv.AddNode(ctx, "anchor"); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
ui := catalogue.Manifest{Module: "ui", Version: "1", Settings: map[string]catalogue.SettingDeclaration{
|
||||
"font-size": {Kind: catalogue.KindPreference, Default: 14, Why: "readable on the desktop"}}}
|
||||
if err := inv.RegisterModule(ctx, ui, inventory.Source{}); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
pass()
|
||||
assignmentWrites, declaredWrites := writes(t, api, Assignments.Bucket), writes(t, api, DeclaredConfig.Bucket)
|
||||
assignmentsSaid, declaredSaid := len(settle(heardAssignments, 0)), len(settle(heardDeclared, 1))
|
||||
|
||||
// **A registration**: a module new to the catalogue.
|
||||
if err := inv.RegisterModule(ctx, catalogue.Manifest{Module: "notes", Version: "1"}, inventory.Source{}); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
ra, rd := pass()
|
||||
if ra != (Result{}) || rd != (Result{Written: 1}) {
|
||||
t.Errorf("a registration wrote assignments %+v and declared config %+v", ra, rd)
|
||||
}
|
||||
if got := writes(t, api, DeclaredConfig.Bucket) - declaredWrites; got != 1 {
|
||||
t.Errorf("a registration changed %d keys of the declared config", got)
|
||||
}
|
||||
if got := writes(t, api, Assignments.Bucket) - assignmentWrites; got != 0 {
|
||||
t.Errorf("a registration changed %d keys of the assignments", got)
|
||||
}
|
||||
said := settle(heardDeclared, declaredSaid+1)[declaredSaid:]
|
||||
if len(said) != 1 || said[0].Key != "notes" || said[0].Change != Added {
|
||||
t.Errorf("a registration raised %+v", said)
|
||||
}
|
||||
declaredWrites, declaredSaid = writes(t, api, DeclaredConfig.Bucket), declaredSaid+len(said)
|
||||
|
||||
// **An assignment.**
|
||||
if _, err := inv.Assign(ctx, "anchor", "ui"); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
ra, rd = pass()
|
||||
if ra != (Result{Written: 1}) || rd != (Result{}) {
|
||||
t.Errorf("an assignment wrote assignments %+v and declared config %+v", ra, rd)
|
||||
}
|
||||
if got := writes(t, api, Assignments.Bucket) - assignmentWrites; got != 1 {
|
||||
t.Errorf("an assignment changed %d keys of the assignments", got)
|
||||
}
|
||||
said = settle(heardAssignments, assignmentsSaid+1)[assignmentsSaid:]
|
||||
if len(said) != 1 || said[0].Key != "anchor.ui" || said[0].Change != Added {
|
||||
t.Errorf("an assignment raised %+v", said)
|
||||
}
|
||||
assignmentWrites, assignmentsSaid = writes(t, api, Assignments.Bucket), assignmentsSaid+len(said)
|
||||
e, err := assignments.KV.Get(ctx, "anchor.ui")
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
var value inventory.AssignmentMirrored
|
||||
if err := json.Unmarshal(e.Value(), &value); err != nil || value.Node != "anchor" || value.Module != "ui" ||
|
||||
value.Assigned.IsZero() {
|
||||
t.Errorf("the assignment is mirrored as %s (%v)", e.Value(), err)
|
||||
}
|
||||
|
||||
// **The same records written again change nothing**: an assignment made again, and a module registered
|
||||
// again with the manifest it had.
|
||||
if _, err := inv.Assign(ctx, "anchor", "ui"); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if err := inv.RegisterModule(ctx, ui, inventory.Source{}); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if ra, rd := pass(); ra != (Result{}) || rd != (Result{}) {
|
||||
t.Errorf("unchanged records wrote assignments %+v and declared config %+v", ra, rd)
|
||||
}
|
||||
if writes(t, api, Assignments.Bucket) != assignmentWrites || writes(t, api, DeclaredConfig.Bucket) != declaredWrites {
|
||||
t.Error("unchanged records changed a key")
|
||||
}
|
||||
if a, d := settle(heardAssignments, assignmentsSaid+1), settle(heardDeclared, declaredSaid+1); len(a) != assignmentsSaid ||
|
||||
len(d) != declaredSaid {
|
||||
t.Errorf("unchanged records raised %+v and %+v", a[assignmentsSaid:], d[declaredSaid:])
|
||||
}
|
||||
|
||||
// **A declaration changed by a registration**: one key, one event, naming the field that moved.
|
||||
ui.Version = "2"
|
||||
ui.Settings["font-size"] = catalogue.SettingDeclaration{Kind: catalogue.KindPreference, Default: 16,
|
||||
Why: "readable on the desktop"}
|
||||
if err := inv.RegisterModule(ctx, ui, inventory.Source{}); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if ra, rd := pass(); ra != (Result{}) || rd != (Result{Written: 1}) {
|
||||
t.Errorf("a changed declaration wrote assignments %+v and declared config %+v", ra, rd)
|
||||
}
|
||||
said = settle(heardDeclared, declaredSaid+1)[declaredSaid:]
|
||||
if len(said) != 1 || said[0].Key != "ui" || said[0].Change != Changed || strings.Join(said[0].Fields, ",") != "settings" {
|
||||
t.Errorf("a changed declaration raised %+v", said)
|
||||
}
|
||||
got, err := declared.KV.Get(ctx, "ui")
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if want := `{"module":"ui","settings":{"font-size":{"kind":"preference","default":16,"why":"readable on the desktop"}}}`; string(got.Value()) != want {
|
||||
t.Errorf("the declaration is mirrored as %s, want %s", got.Value(), want)
|
||||
}
|
||||
|
||||
// **An unassignment** removes its key, said once.
|
||||
if err := inv.Unassign(ctx, "anchor", "ui"); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if ra, _ := pass(); ra != (Result{Removed: 1}) {
|
||||
t.Errorf("an unassignment did %+v", ra)
|
||||
}
|
||||
said = settle(heardAssignments, assignmentsSaid+1)[assignmentsSaid:]
|
||||
if len(said) != 1 || said[0].Key != "anchor.ui" || said[0].Change != Removed {
|
||||
t.Errorf("an unassignment raised %+v", said)
|
||||
}
|
||||
|
||||
// **A module no longer registered** has its key removed, said once as `removed` (ADR 0356 §4).
|
||||
if err := inv.ForgetModule(ctx, "notes"); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
declaredWrites = writes(t, api, DeclaredConfig.Bucket)
|
||||
declaredSaid = len(heardDeclared.all())
|
||||
if _, rd := pass(); rd != (Result{Removed: 1}) {
|
||||
t.Errorf("a module forgotten did %+v", rd)
|
||||
}
|
||||
if got := writes(t, api, DeclaredConfig.Bucket) - declaredWrites; got != 1 {
|
||||
t.Errorf("a module forgotten changed %d keys of the declared config", got)
|
||||
}
|
||||
if _, err := declared.KV.Get(ctx, "notes"); err == nil {
|
||||
t.Error("a module forgotten keeps its key")
|
||||
}
|
||||
said = settle(heardDeclared, declaredSaid+1)[declaredSaid:]
|
||||
if len(said) != 1 || said[0].Key != "notes" || said[0].Change != Removed {
|
||||
t.Errorf("a module forgotten raised %+v", said)
|
||||
}
|
||||
}
|
||||
@@ -1,356 +0,0 @@
|
||||
package recordmirror
|
||||
|
||||
import (
|
||||
"encoding/json"
|
||||
"errors"
|
||||
"slices"
|
||||
"sync"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
"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/link"
|
||||
"github.com/novox/mesh-controller/internal/testbus"
|
||||
)
|
||||
|
||||
// A record mirror against a real bus (novox/hq design 54 §2, §3; ADR 0289): a change of a record changes
|
||||
// exactly one key and raises exactly one event; a pass over unchanged records writes nothing and says
|
||||
// nothing; a record gone is a purge said as `removed`.
|
||||
|
||||
// aMirrorBus is a bus of this test alone with the mesh's streams and the controller's buckets, as the
|
||||
// controller asserts them at its start, and a counter of the writes made to one bucket.
|
||||
func aMirrorBus(t *testing.T) (*broker.JetStream, jetstream.JetStream) {
|
||||
t.Helper()
|
||||
js, err := broker.Dial(testbus.URL(t))
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
t.Cleanup(js.Close)
|
||||
if err := broker.AssertMeshStreams(js); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if err := js.EnsureControllerBuckets(); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
api, err := jetstream.New(js.Conn())
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
return js, api
|
||||
}
|
||||
|
||||
// heard counts what is said on one subject, each body kept.
|
||||
type heard struct {
|
||||
mu sync.Mutex
|
||||
bodies []Event
|
||||
}
|
||||
|
||||
func (h *heard) all() []Event {
|
||||
h.mu.Lock()
|
||||
defer h.mu.Unlock()
|
||||
return append([]Event(nil), h.bodies...)
|
||||
}
|
||||
|
||||
func listen(t *testing.T, conn *nats.Conn, subject string) *heard {
|
||||
t.Helper()
|
||||
h := &heard{}
|
||||
sub, err := conn.Subscribe(subject, func(m *nats.Msg) {
|
||||
var e Event
|
||||
if err := json.Unmarshal(m.Data, &e); err != nil {
|
||||
t.Errorf("an event that is not one: %s", m.Data)
|
||||
return
|
||||
}
|
||||
h.mu.Lock()
|
||||
h.bodies = append(h.bodies, e)
|
||||
h.mu.Unlock()
|
||||
})
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
t.Cleanup(func() { _ = sub.Unsubscribe() })
|
||||
if err := conn.Flush(); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
return h
|
||||
}
|
||||
|
||||
// writes is how many messages the bucket's stream has taken: every put and every purge is one.
|
||||
func writes(t *testing.T, api jetstream.JetStream, bucket string) uint64 {
|
||||
t.Helper()
|
||||
s, err := api.Stream(t.Context(), "KV_"+bucket)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
info, err := s.Info(t.Context())
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
return info.State.LastSeq
|
||||
}
|
||||
|
||||
// settle waits until n events are heard or a second passed, and answers what was heard.
|
||||
func settle(h *heard, n int) []Event {
|
||||
deadline := time.Now().Add(time.Second)
|
||||
for time.Now().Before(deadline) && len(h.all()) < n {
|
||||
time.Sleep(10 * time.Millisecond)
|
||||
}
|
||||
// And a moment more, so an event too many is heard too.
|
||||
time.Sleep(100 * time.Millisecond)
|
||||
return h.all()
|
||||
}
|
||||
|
||||
func aWriter(t *testing.T, js *broker.JetStream, api jetstream.JetStream, m Mirror) *Writer {
|
||||
t.Helper()
|
||||
kv, err := api.KeyValue(t.Context(), m.Bucket)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
return &Writer{Mirror: m, KV: kv, Announce: link.OverNATS{Conn: js.Conn(), JS: js.Context()},
|
||||
Epoch: func() uint64 { return 7 }, Holds: func() bool { return true }}
|
||||
}
|
||||
|
||||
func TestNatsAMirrorChangesOneKeyAndRaisesOneEventPerChange(t *testing.T) {
|
||||
for _, m := range []Mirror{Assignments, DeclaredConfig} {
|
||||
t.Run(m.Name, func(t *testing.T) {
|
||||
js, api := aMirrorBus(t)
|
||||
events := listen(t, js.Conn(), "mesh.seat."+broker.ControllerSeat+".event."+m.Event)
|
||||
w := aWriter(t, js, api, m)
|
||||
ctx := t.Context()
|
||||
|
||||
first := map[string][]byte{"anchor.gitea": []byte(`{"a":1}`)}
|
||||
if _, err := w.Sync(ctx, first); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
before := writes(t, api, m.Bucket)
|
||||
heardBefore := len(settle(events, 1))
|
||||
|
||||
// One record changes, one is added: two keys, two events, nothing else.
|
||||
second := map[string][]byte{"anchor.gitea": []byte(`{"a":2}`), "anchor.forgejo": []byte(`{"a":1}`)}
|
||||
r, err := w.Sync(ctx, second)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if r.Written != 2 || r.Removed != 0 {
|
||||
t.Errorf("a pass over one changed and one new record wrote %+v", r)
|
||||
}
|
||||
if got := writes(t, api, m.Bucket) - before; got != 2 {
|
||||
t.Errorf("the bucket took %d writes for two changed records", got)
|
||||
}
|
||||
said := settle(events, heardBefore+2)[heardBefore:]
|
||||
if len(said) != 2 {
|
||||
t.Fatalf("two changes raised %d events: %+v", len(said), said)
|
||||
}
|
||||
changes := map[string]Event{}
|
||||
for _, e := range said {
|
||||
changes[e.Key] = e
|
||||
}
|
||||
if e := changes["anchor.gitea"]; e.Change != Changed || !slices.Equal(e.Fields, []string{"a"}) ||
|
||||
e.Mirror != m.Name || e.Epoch != 7 || e.Revision == 0 {
|
||||
t.Errorf("the change of anchor.gitea was said as %+v", e)
|
||||
}
|
||||
if e := changes["anchor.forgejo"]; e.Change != Added {
|
||||
t.Errorf("the new anchor.forgejo was said as %+v", e)
|
||||
}
|
||||
|
||||
// **An unchanged pass writes nothing**, and says nothing (ADR 0289 §1).
|
||||
heardBefore = len(events.all())
|
||||
before = writes(t, api, m.Bucket)
|
||||
r, err = w.Sync(ctx, second)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if r != (Result{}) {
|
||||
t.Errorf("a pass over unchanged records did %+v", r)
|
||||
}
|
||||
if got := writes(t, api, m.Bucket) - before; got != 0 {
|
||||
t.Errorf("a pass over unchanged records wrote the bucket %d times", got)
|
||||
}
|
||||
if said := settle(events, heardBefore+1)[heardBefore:]; len(said) != 0 {
|
||||
t.Errorf("a pass over unchanged records raised %+v", said)
|
||||
}
|
||||
|
||||
// **A record gone is a purge**, one write and one `removed`.
|
||||
before = writes(t, api, m.Bucket)
|
||||
r, err = w.Sync(ctx, map[string][]byte{"anchor.gitea": []byte(`{"a":2}`)})
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if r.Removed != 1 || r.Written != 0 {
|
||||
t.Errorf("a record gone did %+v", r)
|
||||
}
|
||||
if got := writes(t, api, m.Bucket) - before; got != 1 {
|
||||
t.Errorf("a record gone wrote the bucket %d times", got)
|
||||
}
|
||||
if _, err := w.KV.Get(ctx, "anchor.forgejo"); !errors.Is(err, jetstream.ErrKeyNotFound) {
|
||||
t.Errorf("the key of a record gone is still read: %v", err)
|
||||
}
|
||||
said = settle(events, heardBefore+1)[heardBefore:]
|
||||
if len(said) != 1 || said[0].Change != Removed || said[0].Key != "anchor.forgejo" {
|
||||
t.Errorf("a record gone was said as %+v", said)
|
||||
}
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
// **A controller started again writes nothing for what the bucket already holds** (design 54 §2: the writer's
|
||||
// memory is filled from the bucket), and a key nobody holds a record for is removed.
|
||||
func TestNatsAWriterStartedAgainWritesOnlyWhatDiffers(t *testing.T) {
|
||||
js, api := aMirrorBus(t)
|
||||
ctx := t.Context()
|
||||
first := aWriter(t, js, api, Assignments)
|
||||
held := map[string][]byte{"anchor.gitea": []byte(`{"a":1}`), "anchor.forgejo": []byte(`{"a":1}`)}
|
||||
if _, err := first.Sync(ctx, held); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
before := writes(t, api, Assignments.Bucket)
|
||||
again := aWriter(t, js, api, Assignments)
|
||||
r, err := again.Sync(ctx, held)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if r != (Result{}) || writes(t, api, Assignments.Bucket) != before {
|
||||
t.Errorf("a writer started again over unchanged records did %+v", r)
|
||||
}
|
||||
}
|
||||
|
||||
// **Another writer in between is no lost value** (design 54 §2, compare-and-set): a write refused because the
|
||||
// key moved reads it again, takes the revision when it already holds the value, and otherwise writes again.
|
||||
func TestNatsAWriteRefusedByCompareAndSetReadsAgain(t *testing.T) {
|
||||
js, api := aMirrorBus(t)
|
||||
ctx := t.Context()
|
||||
w := aWriter(t, js, api, Assignments)
|
||||
if _, err := w.Sync(ctx, map[string][]byte{"anchor.gitea": []byte(`{"a":1}`)}); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
// Behind its back, as a former lease holder would.
|
||||
if _, err := w.KV.Put(ctx, "anchor.gitea", []byte(`{"a":9}`)); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
r, err := w.Sync(ctx, map[string][]byte{"anchor.gitea": []byte(`{"a":2}`)})
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if r.Written != 1 {
|
||||
t.Errorf("a refused write was not made again: %+v", r)
|
||||
}
|
||||
e, err := w.KV.Get(ctx, "anchor.gitea")
|
||||
if err != nil || string(e.Value()) != `{"a":2}` {
|
||||
t.Fatalf("the bucket holds %v, %v", e, err)
|
||||
}
|
||||
}
|
||||
|
||||
// **A value written behind the writer's back is put right on the next pass** (ADR 0289 §10): a former lease
|
||||
// holder's write differs from the store, and the store's value is written again, said as one change.
|
||||
func TestNatsAForeignWriteIsCorrectedOnTheNextPass(t *testing.T) {
|
||||
js, api := aMirrorBus(t)
|
||||
ctx := t.Context()
|
||||
events := listen(t, js.Conn(), "mesh.seat."+broker.ControllerSeat+".event."+Assignments.Event)
|
||||
w := aWriter(t, js, api, Assignments)
|
||||
store := map[string][]byte{"anchor.gitea": []byte(`{"a":1}`)}
|
||||
if _, err := w.Sync(ctx, store); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
heardBefore := len(settle(events, 1))
|
||||
if _, err := w.KV.Put(ctx, "anchor.gitea", []byte(`{"a":9}`)); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
r, err := w.Sync(ctx, store)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if r != (Result{Written: 1}) {
|
||||
t.Errorf("a pass after a foreign write did %+v", r)
|
||||
}
|
||||
e, err := w.KV.Get(ctx, "anchor.gitea")
|
||||
if err != nil || string(e.Value()) != `{"a":1}` {
|
||||
t.Fatalf("the bucket still holds %v, %v", e, err)
|
||||
}
|
||||
if said := settle(events, heardBefore+1)[heardBefore:]; len(said) != 1 || said[0].Change != Changed {
|
||||
t.Errorf("the correction was said as %+v", said)
|
||||
}
|
||||
// And put right, it stays: the next pass writes nothing.
|
||||
if r, err := w.Sync(ctx, store); err != nil || r != (Result{}) {
|
||||
t.Errorf("a pass after the correction did %+v, %v", r, err)
|
||||
}
|
||||
}
|
||||
|
||||
// **A writer without the lease writes nothing** (ADR 0289 §9, design 54 §2): not a new key, not a change, not
|
||||
// a removal, and it says so.
|
||||
func TestNatsAWriterWithoutTheLeaseWritesNothing(t *testing.T) {
|
||||
js, api := aMirrorBus(t)
|
||||
ctx := t.Context()
|
||||
held := aWriter(t, js, api, Assignments)
|
||||
if _, err := held.Sync(ctx, map[string][]byte{"anchor.gitea": []byte(`{"a":1}`)}); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
before := writes(t, api, Assignments.Bucket)
|
||||
for _, holds := range []func() bool{nil, func() bool { return false }} {
|
||||
w := aWriter(t, js, api, Assignments)
|
||||
w.Holds = holds
|
||||
r, err := w.Sync(ctx, map[string][]byte{"anchor.gitea": []byte(`{"a":2}`), "anchor.forgejo": []byte(`{"a":1}`)})
|
||||
if !errors.Is(err, ErrNotHeld) || r != (Result{}) {
|
||||
t.Errorf("a writer without the lease did %+v, %v", r, err)
|
||||
}
|
||||
r, err = w.Sync(ctx, map[string][]byte{})
|
||||
if !errors.Is(err, ErrNotHeld) || r != (Result{}) {
|
||||
t.Errorf("a writer without the lease removed: %+v, %v", r, err)
|
||||
}
|
||||
}
|
||||
// And one that loses it mid-pass stops before its next write.
|
||||
asked := 0
|
||||
w := aWriter(t, js, api, Assignments)
|
||||
w.Holds = func() bool { asked++; return asked <= 2 }
|
||||
r, err := w.Sync(ctx, map[string][]byte{"anchor.a": []byte(`{}`), "anchor.b": []byte(`{}`), "anchor.gitea": []byte(`{"a":1}`)})
|
||||
if !errors.Is(err, ErrNotHeld) || r.Written != 1 {
|
||||
t.Errorf("a writer that lost the lease mid-pass did %+v, %v", r, err)
|
||||
}
|
||||
if got := writes(t, api, Assignments.Bucket) - before; got != 1 {
|
||||
t.Errorf("writers without the lease wrote the bucket %d times", got)
|
||||
}
|
||||
}
|
||||
|
||||
// **A removal asks for the lease too** (ADR 0289 §9): a writer that held the lease when its pass began and lost
|
||||
// it before its first removal purges nothing, and says it does not hold the lease.
|
||||
func TestNatsARemovalWithoutTheLeaseRemovesNothing(t *testing.T) {
|
||||
js, api := aMirrorBus(t)
|
||||
ctx := t.Context()
|
||||
held := aWriter(t, js, api, Assignments)
|
||||
if _, err := held.Sync(ctx, map[string][]byte{"anchor.gitea": []byte(`{"a":1}`)}); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
before := writes(t, api, Assignments.Bucket)
|
||||
asked := 0
|
||||
w := aWriter(t, js, api, Assignments)
|
||||
w.Holds = func() bool { asked++; return asked == 1 }
|
||||
r, err := w.Sync(ctx, map[string][]byte{})
|
||||
if !errors.Is(err, ErrNotHeld) || r != (Result{}) || asked != 2 {
|
||||
t.Errorf("a removal without the lease did %+v, %v (the lease asked %d times)", r, err, asked)
|
||||
}
|
||||
if got := writes(t, api, Assignments.Bucket) - before; got != 0 {
|
||||
t.Errorf("a removal without the lease wrote the bucket %d times", got)
|
||||
}
|
||||
if _, err := w.KV.Get(ctx, "anchor.gitea"); err != nil {
|
||||
t.Errorf("the key was removed without the lease: %v", err)
|
||||
}
|
||||
}
|
||||
|
||||
// **A key the bus cannot hold is refused alone**, and the others are written.
|
||||
func TestAKeyTheBusCannotHoldIsRefusedAlone(t *testing.T) {
|
||||
if validKey("anchor.gitea") != nil || validKey("bad key") == nil || validKey(".lead") == nil ||
|
||||
validKey("") == nil || validKey("trail.") == nil {
|
||||
t.Error("the keys a bucket may hold are not told apart")
|
||||
}
|
||||
}
|
||||
|
||||
func TestTheFieldsThatChangedAreNamed(t *testing.T) {
|
||||
got := changedFields([]byte(`{"a":1,"b":{"x":1},"c":"same"}`), []byte(`{"a":2,"b":{"x":1},"d":true,"c":"same"}`))
|
||||
if !slices.Equal(got, []string{"a", "d"}) {
|
||||
t.Errorf("the fields that changed are %v", got)
|
||||
}
|
||||
if got := changedFields(nil, []byte(`{"a":1}`)); got != nil {
|
||||
t.Errorf("an added key names fields %v", got)
|
||||
}
|
||||
}
|
||||
@@ -1,181 +0,0 @@
|
||||
package recordmirror
|
||||
|
||||
import (
|
||||
"go/ast"
|
||||
"go/parser"
|
||||
"go/token"
|
||||
"io/fs"
|
||||
"os"
|
||||
"path/filepath"
|
||||
"slices"
|
||||
"strconv"
|
||||
"strings"
|
||||
"testing"
|
||||
|
||||
"github.com/novox/mesh-controller/internal/broker"
|
||||
)
|
||||
|
||||
// mayNameAMirror are the files of the controller that may name a record mirror's bucket outside their tests:
|
||||
// where the buckets are defined, asserted and granted (internal/broker), the list a reader names in `reads`
|
||||
// (internal/catalogue/state.go), and each mirror's own writer (this package and the loop that runs it). The
|
||||
// audit of a mirror against its store (ADR 0289 §4, issue 419) joins this list when it is built.
|
||||
var mayNameAMirror = []string{
|
||||
"cmd/mesh-controller/record_mirrors.go",
|
||||
"internal/broker/controller_buckets.go",
|
||||
"internal/broker/writers.go",
|
||||
"internal/catalogue/state.go",
|
||||
}
|
||||
|
||||
// The packages whose names stand for a mirror, by import path, and the names in each.
|
||||
const (
|
||||
brokerPath = "github.com/novox/mesh-controller/internal/broker"
|
||||
recordmirrorPath = "github.com/novox/mesh-controller/internal/recordmirror"
|
||||
)
|
||||
|
||||
var mirrorNames = map[string][]string{
|
||||
brokerPath: {"AssignmentsBucket", "DeclaredConfigBucket", "MirrorAssignments", "MirrorDeclaredConfig",
|
||||
"ControllerMirrors"},
|
||||
recordmirrorPath: {"Assignments", "DeclaredConfig", "Mirrors"},
|
||||
}
|
||||
|
||||
// **Nothing in the controller reads a mirror to decide anything** (ADR 0289 §19 as ADR 0356 §2 changes it): no
|
||||
// file of the controller's packages outside its tests names a mirror's bucket but the audit and a mirror's own
|
||||
// writer. A reader elsewhere fails here, naming the file and what it named.
|
||||
//
|
||||
// **What it catches**: a name that stands for a mirror, qualified by the package it comes from under any
|
||||
// local name the file imports it as (`broker.AssignmentsBucket`, `b.AssignmentsBucket`); the same names bare
|
||||
// in a file of package broker or package recordmirror; and a string literal holding a mirror's bucket name.
|
||||
// **What it does not catch**: a bucket name built at run time, such as `BucketName(ControllerSeat,
|
||||
// "assignments")` or literals concatenated into one, and a mirror reached through a value passed in from an
|
||||
// allowed file. Those are deliberate evasions rather than ordinary code; a reviewer judges them.
|
||||
func TestOnlyTheAuditAndAMirrorsWriterReadAMirror(t *testing.T) {
|
||||
var buckets []string
|
||||
for _, m := range broker.ControllerMirrors() {
|
||||
buckets = append(buckets, broker.BucketName(broker.ControllerSeat, m))
|
||||
}
|
||||
root := filepath.Join("..", "..")
|
||||
found := mirrorReaders(t, root, buckets)
|
||||
if len(found) > 0 {
|
||||
t.Fatalf("a package of the controller reads a record mirror, which only the audit and the mirror's "+
|
||||
"writer may (ADR 0289 §19, ADR 0356 §2):\n %s", strings.Join(found, "\n "))
|
||||
}
|
||||
}
|
||||
|
||||
// **The reader test sees what it says it sees**: in a planted tree, each planted reader is named, and an allowed
|
||||
// file and a selector on a value are not.
|
||||
func TestTheReaderTestFindsPlantedReaders(t *testing.T) {
|
||||
root := t.TempDir()
|
||||
plant := func(path, src string) {
|
||||
full := filepath.Join(root, path)
|
||||
if err := os.MkdirAll(filepath.Dir(full), 0o755); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if err := os.WriteFile(full, []byte(src), 0o644); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
}
|
||||
plant("cmd/mesh-controller/qualified.go", "package main\n\nimport \""+brokerPath+"\"\n\nvar a = broker.AssignmentsBucket\n")
|
||||
plant("internal/x/aliased.go", "package x\n\nimport b \""+brokerPath+"\"\n\nvar a = b.DeclaredConfigBucket\n")
|
||||
plant("internal/broker/zz_reader.go", "package broker\n\nfunc zzR() string { return AssignmentsBucket }\n")
|
||||
plant("internal/inventory/literal.go", "package inventory\n\nvar a = \"mesh-controller_declared-config\"\n")
|
||||
plant("internal/y/writer.go", "package y\n\nimport \""+recordmirrorPath+"\"\n\nvar a = recordmirror.Assignments\n")
|
||||
// Allowed: a mirror's definition and its writer.
|
||||
plant("internal/broker/controller_buckets.go", "package broker\n\nvar AssignmentsBucket = \"x\"\n")
|
||||
plant("internal/broker/other.go", "package broker\n\nfunc other() string { return \"nothing\" }\n")
|
||||
// A selector on a value (v.AssignmentsBucket) is no use of the mirror, but declaring a field of a mirror's
|
||||
// name in package broker is named, on the safe side; and a bare use inside a call under a selector is found.
|
||||
plant("internal/broker/field.go", "package broker\n\ntype s struct{ AssignmentsBucket string }\n\nfunc f(v s) string { return v.AssignmentsBucket }\n")
|
||||
plant("internal/broker/nested.go", "package broker\n\nfunc g() int { return len(h(DeclaredConfigBucket).x) }\n")
|
||||
found := mirrorReaders(t, root, []string{"mesh-controller_assignments", "mesh-controller_declared-config"})
|
||||
want := []string{
|
||||
"cmd/mesh-controller/qualified.go: broker.AssignmentsBucket",
|
||||
"internal/broker/field.go: AssignmentsBucket",
|
||||
"internal/broker/nested.go: DeclaredConfigBucket",
|
||||
"internal/broker/zz_reader.go: AssignmentsBucket",
|
||||
"internal/inventory/literal.go: \"mesh-controller_declared-config\"",
|
||||
"internal/x/aliased.go: b.DeclaredConfigBucket",
|
||||
"internal/y/writer.go: recordmirror.Assignments",
|
||||
}
|
||||
slices.Sort(found)
|
||||
if !slices.Equal(found, want) {
|
||||
t.Errorf("the planted readers were found as\n %s\nwant\n %s", strings.Join(found, "\n "),
|
||||
strings.Join(want, "\n "))
|
||||
}
|
||||
}
|
||||
|
||||
// mirrorReaders is every place under root's cmd and internal, outside tests and the allowed files, that names a
|
||||
// mirror.
|
||||
func mirrorReaders(t *testing.T, root string, buckets []string) []string {
|
||||
t.Helper()
|
||||
var found []string
|
||||
for _, dir := range []string{"cmd", "internal"} {
|
||||
err := filepath.WalkDir(filepath.Join(root, dir), func(path string, e fs.DirEntry, err error) error {
|
||||
if err != nil || e.IsDir() || !strings.HasSuffix(path, ".go") || strings.HasSuffix(path, "_test.go") {
|
||||
return err
|
||||
}
|
||||
rel, err := filepath.Rel(root, path)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
rel = filepath.ToSlash(rel)
|
||||
if strings.HasPrefix(rel, "internal/recordmirror/") || slices.Contains(mayNameAMirror, rel) {
|
||||
return nil
|
||||
}
|
||||
f, err := parser.ParseFile(token.NewFileSet(), path, nil, 0)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
// Each import's local name, mapped to the package it stands for.
|
||||
imported := map[string]string{}
|
||||
for _, imp := range f.Imports {
|
||||
p, _ := strconv.Unquote(imp.Path.Value)
|
||||
if _, ok := mirrorNames[p]; !ok {
|
||||
continue
|
||||
}
|
||||
local := filepath.Base(p)
|
||||
if imp.Name != nil {
|
||||
local = imp.Name.Name
|
||||
}
|
||||
imported[local] = p
|
||||
}
|
||||
// A file of the package that defines the names may use them bare.
|
||||
var bare []string
|
||||
switch f.Name.Name {
|
||||
case "broker":
|
||||
bare = mirrorNames[brokerPath]
|
||||
case "recordmirror":
|
||||
bare = mirrorNames[recordmirrorPath]
|
||||
}
|
||||
ast.Inspect(f, func(n ast.Node) bool {
|
||||
switch n := n.(type) {
|
||||
case *ast.SelectorExpr:
|
||||
pkg, ok := n.X.(*ast.Ident)
|
||||
if !ok {
|
||||
return true
|
||||
}
|
||||
if p, imp := imported[pkg.Name]; imp && slices.Contains(mirrorNames[p], n.Sel.Name) {
|
||||
found = append(found, rel+": "+pkg.Name+"."+n.Sel.Name)
|
||||
}
|
||||
// Neither a selector's own name nor its package is a bare use.
|
||||
return false
|
||||
case *ast.Ident:
|
||||
if slices.Contains(bare, n.Name) {
|
||||
found = append(found, rel+": "+n.Name)
|
||||
}
|
||||
case *ast.BasicLit:
|
||||
for _, b := range buckets {
|
||||
if n.Kind == token.STRING && strings.Contains(n.Value, b) {
|
||||
found = append(found, rel+": "+n.Value)
|
||||
}
|
||||
}
|
||||
}
|
||||
return true
|
||||
})
|
||||
return nil
|
||||
})
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
}
|
||||
return found
|
||||
}
|
||||
Vendored
+1
-1
@@ -8,7 +8,7 @@ private addresses. That is acceptable while mesh-controller is private; if it ev
|
||||
a second place to clean besides the catalogue itself.
|
||||
|
||||
mesh-catalog e0bc919d4945bb5f13f971f74643a0b714cf45da modules/*/module.json, modules/nats/Dockerfile
|
||||
mesh-host c8920622516548978d92e734f3991a262f98a742 examples/foundation-first-node.lock
|
||||
mesh-host 7a41e120635b3f7b6cb94cca8b3fc4093e4e6110 examples/foundation-first-node.lock
|
||||
mesh-sdk 5933eb757d443293631be2bed27a96288c2348d8 conformance/events/module-event.json
|
||||
|
||||
To move them, from this repository's root, with the two repositories checked out beside it:
|
||||
|
||||
+29
-79
@@ -28,12 +28,11 @@
|
||||
// image, and this file is applied on a machine with no mesh to ask about anything (ADR 0006). The
|
||||
// registry prefix is the lab's; the digests are upstream's.
|
||||
//
|
||||
// **The store is raised where the `postgres` module keeps it** (novox/hq ADR 0338): the container
|
||||
// `postgres`, on the module's pgvector image, its data in the directory /var/lib/mesh-store owned by
|
||||
// the image's postgres user, its superuser password read from the module's own-secret file — so the
|
||||
// store step adopts it as it runs. Pinned upstream by the module's own digest, not through the lab's
|
||||
// registry prefix, because the two must be the same reference (mesh-host internal/bootstrap holds
|
||||
// them to each other against the module's manifest).
|
||||
// The store's data is a NAMED VOLUME, not a directory on the machine. A directory the host
|
||||
// creates is owned by root, and the database runs as somebody else inside the container — so it
|
||||
// could not write, and the container crash-looped. A named volume lets the image set up its own
|
||||
// ownership, and outlives the container, which is what you want for the thing holding the mesh's
|
||||
// state.
|
||||
|
||||
{
|
||||
"declaration": 1,
|
||||
@@ -84,28 +83,17 @@
|
||||
"boot": "enabled",
|
||||
"restart-on": ["base-filter"]
|
||||
},
|
||||
// The store's data is a directory owned by the image's postgres user (999), made before the
|
||||
// container starts: a directory the runtime made would be root's, and the database could not
|
||||
// write it. Measured on the module's image 2026-10-13: it initialises a new store on a directory
|
||||
// made so (novox/hq ADR 0338 §5).
|
||||
{
|
||||
"id": "store-data",
|
||||
"type": "directory",
|
||||
"path": "/var/lib/mesh-store",
|
||||
"mode": "0700",
|
||||
"owner": "999:70"
|
||||
},
|
||||
{
|
||||
"id": "store",
|
||||
"type": "container",
|
||||
"name": "postgres",
|
||||
"image": "pgvector/pgvector@sha256:cf134a767f474095eeba57e0117be8e568e011a63f33fbf252f14c9b760f8e6f",
|
||||
"name": "mesh-store",
|
||||
"image": "192.0.2.250:5000/postgres@sha256:7abf537131b66ed5af448d90653abf1679b0c7e9a1f07efdd4c3108a401b259a",
|
||||
"env": {
|
||||
"POSTGRES_PASSWORD": "bootstrap",
|
||||
"PGDATA": "/var/lib/postgresql/data/pgdata"
|
||||
},
|
||||
"ports": ["5432:5432"],
|
||||
"volumes": ["/var/lib/mesh-store:/var/lib/postgresql/data"]
|
||||
"volumes": ["mesh-store-data:/var/lib/postgresql/data"]
|
||||
},
|
||||
// Over TCP, not the socket. While the store initialises it runs a temporary server on the
|
||||
// socket ONLY, then stops it and starts the real one — so a socket check passes, the action
|
||||
@@ -114,21 +102,21 @@
|
||||
{
|
||||
"id": "store-ready",
|
||||
"type": "action",
|
||||
"in": "postgres",
|
||||
"in": "mesh-store",
|
||||
"command": ["sh", "-c", "for i in $(seq 1 180); do pg_isready -h 127.0.0.1 -U postgres >/dev/null 2>&1 && exit 0; sleep 1; done; echo 'the store did not answer within 180s; its own last words follow'; pg_isready -h 127.0.0.1 -U postgres; tail -n 20 /var/lib/postgresql/data/log/*.log 2>/dev/null; exit 1"],
|
||||
"verify": ["pg_isready", "-h", "127.0.0.1", "-U", "postgres"]
|
||||
},
|
||||
{
|
||||
"id": "inventory-database",
|
||||
"type": "action",
|
||||
"in": "postgres",
|
||||
"in": "mesh-store",
|
||||
"command": ["sh", "-c", "psql -U postgres -c 'CREATE DATABASE inventory'"],
|
||||
"verify": ["sh", "-c", "psql -U postgres -lqt | cut -d'|' -f1 | grep -qw inventory"]
|
||||
},
|
||||
{
|
||||
"id": "identity-database",
|
||||
"type": "action",
|
||||
"in": "postgres",
|
||||
"in": "mesh-store",
|
||||
"command": ["sh", "-c", "psql -U postgres -c 'CREATE DATABASE identity'"],
|
||||
"verify": ["sh", "-c", "psql -U postgres -lqt | cut -d'|' -f1 | grep -qw identity"]
|
||||
},
|
||||
@@ -138,44 +126,20 @@
|
||||
{
|
||||
"id": "licences-database",
|
||||
"type": "action",
|
||||
"in": "postgres",
|
||||
"in": "mesh-store",
|
||||
"command": ["sh", "-c", "psql -U postgres -c 'CREATE DATABASE licences'"],
|
||||
"verify": ["sh", "-c", "psql -U postgres -lqt | cut -d'|' -f1 | grep -qw licences"]
|
||||
},
|
||||
{
|
||||
"id": "context-schemas",
|
||||
"type": "action",
|
||||
"command": ["docker", "run", "--rm", "--network", "container:postgres",
|
||||
"command": ["docker", "run", "--rm", "--network", "container:mesh-store",
|
||||
"-e", "MESH_STORE_INVENTORY=postgres://postgres:bootstrap@127.0.0.1:5432/inventory?sslmode=disable",
|
||||
"-e", "MESH_STORE_IDENTITY=postgres://postgres:bootstrap@127.0.0.1:5432/identity?sslmode=disable",
|
||||
"-e", "MESH_STORE_LICENCES=postgres://postgres:bootstrap@127.0.0.1:5432/licences?sslmode=disable",
|
||||
"192.0.2.250:5000/mesh-controller@sha256:c67db38439ff0aee242b467486765467bb95801f52175fc5727cc4e437338ace",
|
||||
"migrate"],
|
||||
"verify": ["sh", "-c", "docker exec postgres psql -U postgres -d inventory -tAc \"select to_regclass('public.node')\" | grep -qx node && docker exec postgres psql -U postgres -d identity -tAc \"select to_regclass('public.signing_key')\" | grep -qx signing_key && docker exec postgres psql -U postgres -d licences -tAc \"select to_regclass('public.licence')\" | grep -qx licence"]
|
||||
},
|
||||
// **The bus is raised where the `nats` module keeps it** (novox/hq ADR 0333 §1): the container
|
||||
// `mesh-broker-nats`, its configuration and user list in /var/lib/nats-module/conf, its streams in
|
||||
// /var/lib/mesh-broker-nats and its certificate in /var/lib/mesh-broker-tls — the names and paths the
|
||||
// catalogue's module declares. So adopting it as that module at the broker step keeps it: the
|
||||
// module's built image takes the place of the upstream one below, once, on the same streams, users
|
||||
// and certificate, and nothing else moves. mesh-host's internal/bootstrap holds the two to each other
|
||||
// against the module's own manifest (busLayout), at the broker step and in its tests.
|
||||
//
|
||||
// Directories on the machine, not volumes the runtime names: the module mounts these paths, and the
|
||||
// control plane, once it runs as the mesh's process, reads the certificate at its path on the machine.
|
||||
// The certificate's directory is readable by everyone because the certificate is public and the control
|
||||
// plane runs as nobody; its key is written 0600 by the program that makes it.
|
||||
{
|
||||
"id": "bus-tls-dir",
|
||||
"type": "directory",
|
||||
"path": "/var/lib/mesh-broker-tls",
|
||||
"mode": "0755"
|
||||
},
|
||||
{
|
||||
"id": "bus-data-dir",
|
||||
"type": "directory",
|
||||
"path": "/var/lib/mesh-broker-nats",
|
||||
"mode": "0700"
|
||||
"verify": ["sh", "-c", "docker exec mesh-store psql -U postgres -d inventory -tAc \"select to_regclass('public.node')\" | grep -qx node && docker exec mesh-store psql -U postgres -d identity -tAc \"select to_regclass('public.signing_key')\" | grep -qx signing_key && docker exec mesh-store psql -U postgres -d licences -tAc \"select to_regclass('public.licence')\" | grep -qx licence"]
|
||||
},
|
||||
{
|
||||
"id": "bus-certificate",
|
||||
@@ -187,63 +151,49 @@
|
||||
// it, and asking nothing of the image it writes into. Self-signed on purpose: a host pins
|
||||
// this server's exact certificate (novox/hq ADR 0004), and at this moment there is no mesh
|
||||
// to ask an authority of.
|
||||
// `--user 0:0` because the directory is root's and this image runs as nobody, which is right
|
||||
// for the long-running control plane and wrong for a one-shot writing into a fresh directory.
|
||||
"command": ["docker", "run", "--rm", "--user", "0:0", "-v", "/var/lib/mesh-broker-tls:/tls",
|
||||
// `--user 0:0` because the volume is root's and this image runs as nobody, which is right
|
||||
// for the long-running control plane and wrong for a one-shot writing into a fresh volume.
|
||||
"command": ["docker", "run", "--rm", "--user", "0:0", "-v", "mesh-broker-tls:/tls",
|
||||
"192.0.2.250:5000/mesh-controller@sha256:c67db38439ff0aee242b467486765467bb95801f52175fc5727cc4e437338ace",
|
||||
"broker", "certificate", "--into", "/tls"],
|
||||
"verify": ["docker", "run", "--rm", "--user", "0:0", "-v", "/var/lib/mesh-broker-tls:/tls",
|
||||
"verify": ["docker", "run", "--rm", "--user", "0:0", "-v", "mesh-broker-tls:/tls",
|
||||
"192.0.2.250:5000/mesh-controller@sha256:c67db38439ff0aee242b467486765467bb95801f52175fc5727cc4e437338ace",
|
||||
"broker", "certificate", "--check", "--into", "/tls"]
|
||||
},
|
||||
// **Seeds, written only where nothing is** (create-once). The bus's configuration and user list are
|
||||
// the bundle's until the bus is adopted, and the module's from then on (ADR 0333 §4): an installer
|
||||
// run again after the adoption applies this bundle again, and an ordinary file would put the
|
||||
// bundle's one user back over every user the mesh composed, and leave the node unable to reach its
|
||||
// bus. Until the adoption the installer places the composed list over the seed (enrol.go).
|
||||
{
|
||||
"id": "bus-conf-dir",
|
||||
"type": "directory",
|
||||
"path": "/var/lib/nats-module/conf",
|
||||
"path": "/var/lib/mesh-bus-conf",
|
||||
"mode": "0700"
|
||||
},
|
||||
{
|
||||
"id": "bus-conf",
|
||||
"type": "file",
|
||||
"path": "/var/lib/nats-module/conf/nats.conf",
|
||||
"path": "/var/lib/mesh-bus-conf/nats.conf",
|
||||
"mode": "0644",
|
||||
"create-once": true,
|
||||
"content": "port: 4222\nhttp: 0.0.0.0:8222\n\ntls {\n cert_file: \"/tls/tls.crt\"\n key_file: \"/tls/tls.key\"\n}\n\njetstream {\n store_dir: \"/data\"\n}\n\ninclude accounts.conf\n"
|
||||
"content": "port: 4222\nhttp: 127.0.0.1:8222\n\ntls {\n cert_file: \"/tls/tls.crt\"\n key_file: \"/tls/tls.key\"\n}\n\njetstream {\n store_dir: \"/data\"\n}\n\ninclude accounts.conf\n"
|
||||
},
|
||||
{
|
||||
"id": "bus-accounts",
|
||||
"type": "file",
|
||||
"path": "/var/lib/nats-module/conf/accounts.conf",
|
||||
"path": "/var/lib/mesh-bus-conf/accounts.conf",
|
||||
"mode": "0600",
|
||||
"create-once": true,
|
||||
"content": "// The first user list, carried by the installer because at genesis there is no mesh to\n// compose one. A bootstrap credential, rotated with the store's and replaced by the\n// controller's own composition from its first start onward.\naccounts {\n MESH {\n jetstream: enabled\n users = [\n { user: \"controller\", password: \"$2a$10$AHqJgOifIVbU41KmATiMhuXFs8xa7Wl2HuN4UVBCXdN2jIQzjqApy\", permissions: {\n publish: { allow: [\"$JS.ACK.CONTROL.controller.>\", \"$JS.ACK.EVENTS.controller.>\", \"$JS.ACK.DEAD_LETTER_NOTICES.controller.>\", \"$JS.API.>\", \"$KV.mesh-controller_asked.>\", \"$KV.mesh-controller_assignments.>\", \"$KV.mesh-controller_calls.>\", \"$KV.mesh-controller_hand-acts.>\", \"$KV.mesh-controller_conditions.>\", \"$KV.mesh-controller_condition-history.>\", \"$KV.mesh-controller_declared-config.>\", \"$KV.mesh-controller_lease.>\", \"$KV.SEAT_MESH_BUILD_MACHINE_cancelled.>\", \"$KV.SEAT_NODE_BUILD_AGENT_cancelled.>\", \"_INBOX.enrol.>\", \"mesh.assignment.>\", \"mesh.mod.*.tool.>\", \"mesh.node.>\", \"mesh.seat.mesh-build-machine.accept.>\", \"mesh.seat.node-build-agent.accept.>\", \"mesh.seat.mesh-build-machine.tool.>\", \"mesh.seat.node-build-agent.tool.>\", \"mesh.seat.mesh-controller.event.applied\", \"mesh.seat.mesh-controller.event.built-before\", \"mesh.seat.mesh-controller.event.refused\", \"mesh.seat.mesh-controller.event.rolled-back\", \"mesh.seat.mesh-controller.event.checked\", \"mesh.seat.mesh-controller.event.plan-moved\", \"mesh.seat.mesh-controller.event.assignment-changed\", \"mesh.seat.mesh-controller.event.declared-config-changed\", \"mesh.seat.mesh-controller.event.condition-raised\", \"mesh.seat.mesh-controller.event.condition-changed\", \"mesh.seat.mesh-controller.event.condition-cleared\", \"mesh.seat.mesh-controller.event.doctor-heartbeat\", \"mesh.seat.mesh-controller.event.secret-replaced\", \"mesh.seat.mesh-controller.event.healer-acted\", \"$SRV.INFO\", \"mesh.seat.node-intrusion-prevention.tool.banned.*\", \"mesh.seat.node-backup.tool.backed-up.*\", \"mesh.seat.node-backup.tool.now.*\", \"mesh.seat.mesh-delivery.tool.stalled\", \"mesh.seat.mesh-delivery.tool.close\", \"mesh.seat.node-launcher.tool.secret.*\", \"mesh.seat.mesh-delivery.tool.release\", \"mesh.seat.mesh-delivery.tool.stop\", \"mesh.seat.mesh-delivery.tool.times\", \"mesh.seat.mesh-delivery.tool.groups\", \"mesh.seat.node-service-manager.tool.restart.*\", \"mesh.seat.mesh-controller.tool.plans\", \"mesh.events.dead.>\", \"mesh.again.>\"] }\n subscribe: { allow: [\"$JS.API.>\", \"$JS.EVENT.ADVISORY.CONSUMER.MAX_DELIVERIES.>\", \"$JS.EVENT.ADVISORY.CONSUMER.DELETED.>\", \"_DELIVER.controller\", \"_DELIVER.controller.>\", \"_INBOX.controller.>\", \"mesh.control.>\", \"mesh.mod.*.event.provisioner.failing\", \"mesh.mod.*.event.provisioner.recovered\", \"mesh.mod.*.event.provisioner.retirement\", \"mesh.mod.gitea.event.pull.merged\", \"mesh.mod.gitea.event.pull.updated\", \"mesh.mod.mesh-catalog.event.catching-up\", \"mesh.mod.mesh-catalog.event.upgraded\", \"mesh.seat.mesh-build-machine.event.built\", \"mesh.seat.node-build-agent.event.built\", \"mesh.seat.operator-channel.event.decided.mesh-controller\", \"mesh.seat.mesh-controller.tool.>\", \"$SRV.PING\", \"$SRV.INFO\", \"$SRV.PING.mesh-controller\", \"$SRV.PING.mesh-controller.>\", \"$SRV.INFO.mesh-controller\", \"$SRV.INFO.mesh-controller.>\", \"$SRV.STATS\", \"$SRV.STATS.mesh-controller\", \"$SRV.STATS.mesh-controller.>\"] }\n allow_responses: { max: 1, ttl: \"1m\" }\n } }\n ]\n }\n}\n"
|
||||
"content": "// The first user list, carried by the installer because at genesis there is no mesh to\n// compose one. A bootstrap credential, rotated with the store's and replaced by the\n// controller's own composition from its first start onward.\naccounts {\n MESH {\n jetstream: enabled\n users = [\n { user: \"controller\", password: \"$2a$10$AHqJgOifIVbU41KmATiMhuXFs8xa7Wl2HuN4UVBCXdN2jIQzjqApy\", permissions: {\n publish: { allow: [\"$JS.ACK.CONTROL.controller.>\", \"$JS.ACK.EVENTS.controller.>\", \"$JS.ACK.DEAD_LETTER_NOTICES.controller.>\", \"$JS.API.>\", \"$KV.mesh-controller_asked.>\", \"$KV.mesh-controller_calls.>\", \"$KV.mesh-controller_hand-acts.>\", \"$KV.mesh-controller_conditions.>\", \"$KV.mesh-controller_condition-history.>\", \"$KV.mesh-controller_lease.>\", \"$KV.SEAT_MESH_BUILD_MACHINE_cancelled.>\", \"$KV.SEAT_NODE_BUILD_AGENT_cancelled.>\", \"_INBOX.enrol.>\", \"mesh.assignment.>\", \"mesh.mod.*.tool.>\", \"mesh.node.>\", \"mesh.seat.mesh-build-machine.accept.>\", \"mesh.seat.node-build-agent.accept.>\", \"mesh.seat.mesh-build-machine.tool.>\", \"mesh.seat.node-build-agent.tool.>\", \"mesh.seat.mesh-controller.event.applied\", \"mesh.seat.mesh-controller.event.built-before\", \"mesh.seat.mesh-controller.event.refused\", \"mesh.seat.mesh-controller.event.rolled-back\", \"mesh.seat.mesh-controller.event.checked\", \"mesh.seat.mesh-controller.event.plan-moved\", \"mesh.seat.mesh-controller.event.condition-raised\", \"mesh.seat.mesh-controller.event.condition-changed\", \"mesh.seat.mesh-controller.event.condition-cleared\", \"mesh.seat.mesh-controller.event.doctor-heartbeat\", \"mesh.seat.mesh-controller.event.secret-replaced\", \"mesh.seat.mesh-controller.event.healer-acted\", \"$SRV.INFO\", \"mesh.seat.node-intrusion-prevention.tool.banned.*\", \"mesh.seat.node-backup.tool.backed-up.*\", \"mesh.seat.node-backup.tool.now.*\", \"mesh.seat.mesh-delivery.tool.stalled\", \"mesh.seat.mesh-delivery.tool.close\", \"mesh.seat.node-launcher.tool.secret.*\", \"mesh.seat.mesh-delivery.tool.release\", \"mesh.seat.mesh-delivery.tool.stop\", \"mesh.seat.mesh-delivery.tool.times\", \"mesh.seat.mesh-delivery.tool.groups\", \"mesh.seat.node-service-manager.tool.restart.*\", \"mesh.seat.mesh-controller.tool.plans\", \"mesh.events.dead.>\", \"mesh.again.>\"] }\n subscribe: { allow: [\"$JS.API.>\", \"$JS.EVENT.ADVISORY.CONSUMER.MAX_DELIVERIES.>\", \"$JS.EVENT.ADVISORY.CONSUMER.DELETED.>\", \"_DELIVER.controller\", \"_DELIVER.controller.>\", \"_INBOX.controller.>\", \"mesh.control.>\", \"mesh.mod.*.event.provisioner.failing\", \"mesh.mod.*.event.provisioner.recovered\", \"mesh.mod.*.event.provisioner.retirement\", \"mesh.mod.gitea.event.pull.merged\", \"mesh.mod.gitea.event.pull.updated\", \"mesh.mod.mesh-catalog.event.catching-up\", \"mesh.mod.mesh-catalog.event.upgraded\", \"mesh.seat.mesh-build-machine.event.built\", \"mesh.seat.node-build-agent.event.built\", \"mesh.seat.operator-channel.event.decided.mesh-controller\", \"mesh.seat.mesh-controller.tool.>\", \"$SRV.PING\", \"$SRV.INFO\", \"$SRV.PING.mesh-controller\", \"$SRV.PING.mesh-controller.>\", \"$SRV.INFO.mesh-controller\", \"$SRV.INFO.mesh-controller.>\", \"$SRV.STATS\", \"$SRV.STATS.mesh-controller\", \"$SRV.STATS.mesh-controller.>\"] }\n allow_responses: { max: 1, ttl: \"1m\" }\n } }\n ]\n }\n}\n"
|
||||
},
|
||||
// The upstream server, until the broker step adopts it: the `nats` module's own image, built by the
|
||||
// mesh, takes its place there once, on these same directories (ADR 0333 §3), as the bus's planned
|
||||
// step (ADR 0236). The release is the one the module's image is built on (its NATS_BASE, 2.11.17),
|
||||
// pinned upstream by that same digest: the adoption changes no server release, so it can be undone
|
||||
// by putting the old image back and is said reversible. The published port is the
|
||||
// node's bus port, 5671 unless given, which every enrolment token names; the installer gives the
|
||||
// module the setting that maps its declared 4222 to it (ADR 0333 §2).
|
||||
{
|
||||
"id": "broker",
|
||||
"type": "container",
|
||||
"name": "mesh-broker-nats",
|
||||
"image": "nats@sha256:e4bf19f15fd3218814a4e3c9e0064e1334bd8aa20d5984b9f1a0afd084f8cc00",
|
||||
"name": "mesh-broker",
|
||||
"image": "192.0.2.250:5000/nats@sha256:b83efabe3e7def1e0a4a31ec6e078999bb17c80363f881df35edc70fcb6bb927",
|
||||
"ports": ["5671:4222", "127.0.0.1:8222:8222"],
|
||||
"volumes": ["/var/lib/mesh-broker-nats:/data", "/var/lib/nats-module/conf:/etc/nats:ro", "/var/lib/mesh-broker-tls:/tls:ro"],
|
||||
"volumes": ["mesh-broker-data:/data", "mesh-broker-tls:/tls:ro", "/var/lib/mesh-bus-conf:/etc/nats:ro"],
|
||||
"args": ["-c", "/etc/nats/nats.conf", "-js"]
|
||||
},
|
||||
{
|
||||
"id": "broker-ready",
|
||||
"type": "action",
|
||||
"command": ["sh", "-c", "for i in $(seq 1 60); do docker run --rm --network host --entrypoint sh nats@sha256:e4bf19f15fd3218814a4e3c9e0064e1334bd8aa20d5984b9f1a0afd084f8cc00 -c 'nc -z 127.0.0.1 5671' >/dev/null 2>&1 && exit 0; sleep 1; done; exit 1"],
|
||||
"verify": ["sh", "-c", "docker run --rm --network host --entrypoint sh nats@sha256:e4bf19f15fd3218814a4e3c9e0064e1334bd8aa20d5984b9f1a0afd084f8cc00 -c 'nc -z 127.0.0.1 5671'"]
|
||||
"command": ["sh", "-c", "for i in $(seq 1 60); do docker run --rm --network host --entrypoint sh 192.0.2.250:5000/nats@sha256:b83efabe3e7def1e0a4a31ec6e078999bb17c80363f881df35edc70fcb6bb927 -c 'nc -z 127.0.0.1 5671' >/dev/null 2>&1 && exit 0; sleep 1; done; exit 1"],
|
||||
"verify": ["sh", "-c", "docker run --rm --network host --entrypoint sh 192.0.2.250:5000/nats@sha256:b83efabe3e7def1e0a4a31ec6e078999bb17c80363f881df35edc70fcb6bb927 -c 'nc -z 127.0.0.1 5671'"]
|
||||
},
|
||||
{
|
||||
"id": "control-plane",
|
||||
@@ -252,7 +202,7 @@
|
||||
"image": "192.0.2.250:5000/mesh-controller@sha256:c67db38439ff0aee242b467486765467bb95801f52175fc5727cc4e437338ace",
|
||||
"network": "host",
|
||||
"args": ["serve"],
|
||||
"volumes": ["/var/lib/mesh-broker-tls:/broker-tls:ro"],
|
||||
"volumes": ["mesh-broker-tls:/broker-tls:ro"],
|
||||
"env": {
|
||||
"MESH_STORE_INVENTORY": "postgres://postgres:bootstrap@127.0.0.1:5432/inventory?sslmode=disable",
|
||||
"MESH_STORE_IDENTITY": "postgres://postgres:bootstrap@127.0.0.1:5432/identity?sslmode=disable",
|
||||
|
||||
Vendored
-2
@@ -49,8 +49,6 @@
|
||||
./internal/inventory TestACredentialIssuedOutsideIsNeverMade
|
||||
./internal/inventory TestOneAcceptForAProviderIsSealedToEveryConsumer
|
||||
./internal/inventory TestASharedOwnSecretIssuedOutsideIsNeverMade
|
||||
./internal/recordmirror TestNatsAnAssignmentAndARegistrationEachChangeOneKeyAndRaiseOneEvent
|
||||
./internal/inventory TestAModuleInTheConfigFormIsMirroredWithItsWholeDeclaration
|
||||
./cmd/mesh-controller TestOneAcceptForAProviderIsPlacedForEveryConsumer
|
||||
./cmd/mesh-controller TestReplay164
|
||||
./cmd/mesh-controller TestAnApprovedLayerOpensTheWalkThatSendsIt
|
||||
|
||||
Reference in New Issue
Block a user