Land feature 57's SDK work on its release branch
mesh/merge-gate head d3e182503c19 merged into trunk c8b07242a009 — pass: [release feat/57-config-and-provisioning, judged by its controller fcdecb436117] t…
mesh/repo-check head d3e182503c19 merged into trunk c8b07242a009 — pass: [release feat/57-config-and-provisioning, judged by its controller fcdecb436117] i…
mesh/delivery delivered

Providers, the state bus, templates and the environment collection that the
feature's modules build on (feature 57, config and provisioning).
This commit is contained in:
2026-10-15 21:07:29 +00:00
parent c8b07242a0
commit d3e182503c
16 changed files with 1888 additions and 38 deletions
+72
View File
@@ -0,0 +1,72 @@
package provider
import (
"context"
"encoding/json"
"errors"
"fmt"
"io/fs"
"os"
)
// The provider's consumers' credentials, from the file its machine's provisioner client fills (novox/hq ADR 0373,
// issue 679).
//
// **The provider never asks the vault.** A provider in the tool runner has no account of its own, and the vault
// hands a secret only to a node's provisioner client or to a module under an account of its own (ADR 0349 clause
// 11, ADR 0370 clause 2). So the provider's template names `${consumers:<provision>}`, and its machine's client
// fills that file with one entry per consumer bound to it for that provision, each credential asked of the vault in
// the client's own name; the file is the provider's alone (mode 0600, owned). ConsumersFile is that file as the
// helper's Vault: it reads the file on each ask, so a consumer that binds or a credential the vault remakes reaches
// the provider at its next pass.
// ConsumerCredential is one entry of the file: the consumer, the vault's name of its credential, and its value.
type ConsumerCredential struct {
// Consumer and Module are the consumer's machine and module, Local its local name for the provision, if any.
Consumer string `json:"consumer"`
Module string `json:"module"`
Local string `json:"local,omitempty"`
// Credential is the vault's name of the consumer's credential, PairName's.
Credential string `json:"credential"`
// Value is the credential itself.
Value string `json:"value"`
}
// ConsumersFile answers a credential from the files the provider machine's client fills: one per provision whose
// template names `${consumers:<provision>}`, each a JSON list of ConsumerCredential.
type ConsumersFile struct {
Paths []string
}
// Value is the credential named name, from the first file carrying it. A file not there yet is passed over; one not
// in the client's form is an error, never read as no consumers.
func (f ConsumersFile) Value(_ context.Context, name string) ([]byte, error) {
var missing []string
for _, at := range f.Paths {
raw, err := os.ReadFile(at)
if errors.Is(err, fs.ErrNotExist) {
missing = append(missing, at)
continue
}
if err != nil {
return nil, fmt.Errorf("the consumers' file %s: %w", at, err)
}
var entries []ConsumerCredential
if err := json.Unmarshal(raw, &entries); err != nil {
return nil, fmt.Errorf("the consumers' file %s is not the provisioner client's list of consumers", at)
}
for _, e := range entries {
if e.Credential == name {
if e.Value == "" {
return nil, fmt.Errorf("the consumers' file %s carries %s with no value", at, name)
}
return []byte(e.Value), nil
}
}
}
if len(missing) == len(f.Paths) {
return nil, fmt.Errorf("%s: no consumers' file is filled yet (%v)", name, f.Paths)
}
return nil, fmt.Errorf("%s is in no consumers' file: the vault holds none for it yet, or its machine's client "+
"has not filled it", name)
}
+69
View File
@@ -0,0 +1,69 @@
package provider
import (
"context"
"os"
"path/filepath"
"strings"
"testing"
)
// The provider reads its consumers' credentials only from the file its machine's provisioner client fills (novox/hq
// ADR 0373 clause 3, issue 679): one entry per consumer, naming the consumer and the vault's name for its credential,
// and carrying the value. The helper is wired to it as its Vault and asks nothing of the vault itself.
func TestTheConsumersFileAnswersEachConsumersCredential(t *testing.T) {
dir := t.TempDir()
at := filepath.Join(dir, "consumers.json")
file := `[{"consumer":"ace","module":"letta","credential":"pair/novox/postgres-database/ace/letta","value":"s3cret"},
{"consumer":"g14","module":"gitea","local":"cache","credential":"pair/novox/postgres-database/g14/gitea/cache","value":"other"}]`
if err := os.WriteFile(at, []byte(file), 0o600); err != nil {
t.Fatal(err)
}
v := ConsumersFile{Paths: []string{filepath.Join(dir, "absent.json"), at}}
got, err := v.Value(context.Background(), PairName("novox", "postgres-database", "ace", "letta", ""))
if err != nil || string(got) != "s3cret" {
t.Fatalf("ace's letta: %q, %v; want s3cret", got, err)
}
got, err = v.Value(context.Background(), PairName("novox", "postgres-database", "g14", "gitea", "cache"))
if err != nil || string(got) != "other" {
t.Fatalf("g14's gitea under cache: %q, %v; want other", got, err)
}
// A consumer the file does not carry is waited for, and never told of a login its provider has not made.
_, err = v.Value(context.Background(), PairName("novox", "postgres-database", "shanks", "photos", ""))
if err == nil || !strings.Contains(err.Error(), "pair/novox/postgres-database/shanks/photos") {
t.Fatalf("a consumer the file does not carry: %v", err)
}
if strings.Contains(err.Error(), "s3cret") || strings.Contains(err.Error(), "other") {
t.Fatalf("an error names a credential: %v", err)
}
// Wired as the helper's Vault, the file is what makes each consumer's login.
bus := newStandIn()
bus.Put(context.Background(), AssignmentsBucket, "ace.letta", []byte(`{"node":"ace","module":"letta",
"provisioning":{"provisions":[{"provision":"postgres-database","provider":"novox","provider-module":"postgres"}]}}`))
var made []string
p := New(Options{Node: "novox", Module: "postgres", Bus: bus, Vault: v,
Offers: map[string]Offer{"postgres-database": {Serves: map[string]any{"port": 5432}}},
Make: func(_ context.Context, c Consumer) error {
made = append(made, c.Entry.User+"="+string(c.Credential))
return nil
}, Logf: func(string, ...any) {}})
if _, err := p.Pass(context.Background()); err != nil {
t.Fatal(err)
}
if len(made) != 1 || made[0] != "mesh_ace_letta=s3cret" {
t.Fatalf("made %v, want ace's letta under the file's credential", made)
}
}
// A file that is not the client's form is said, never read as no consumers.
func TestAConsumersFileNotInItsFormIsRefused(t *testing.T) {
at := filepath.Join(t.TempDir(), "consumers.json")
if err := os.WriteFile(at, []byte(`{"not": "a list"}`), 0o600); err != nil {
t.Fatal(err)
}
_, err := ConsumersFile{Paths: []string{at}}.Value(context.Background(), "pair/novox/p/ace/m")
if err == nil || !strings.Contains(err.Error(), at) {
t.Fatalf("a file not in the client's form: %v", err)
}
}
+635
View File
@@ -0,0 +1,635 @@
// Package provider is a providing module's half of a binding: it answers the provider's consumers itself, over the
// bus (novox/hq ADR 0349 clause 12, design 57 B1, issue 678).
//
// **Who binds the provider.** Each node's provisioner client records the needs of the modules handed to it from
// their plans, and the controller names in each plan, per required provision and local name, the provider that
// resolution chose: `provisioning.provisions` in `<node>.<module>` of the assignments mirror (issue 760). A recorded
// need names no provider (issue 677's note 7), so the plan's entry is where a bind is seen, as the vault sees it
// (mesh-vault pair.go). A consumer binds this provider when an entry names this machine and this module, and
// unbinds when the entry, or the assignment, goes. A consumer still filled the old way has no plan and is not seen
// here; the controller answers it as before (ADR 0349 clause 17).
//
// **One entry per consumer, in the provider's own bucket.** For each bind the helper composes the consumer's
// served values from the provider's offer (its config already composed into it, ADR 0202 §7) and the consumer's
// identity (`${consumer:as}`, `${consumer:as:dns}`, ADR 0202 §1), and writes one entry, `<node>.<module>.
// <provision>[.<local>]` of `<provider module>_bound`: the one statement both ends read. It never carries a
// secret. It names the credential the vault keeps for the pair, which the consumer's client asks the vault for.
//
// **The credential is the vault's** (ADR 0349 clause 8). The helper mints nothing and reads nothing of the
// provider's own secrets: it reads the value from the vault and hands it to the provider's adapter, which makes the
// login its software keeps. An offer whose software holds one credential (ADR 0158) is the vault's one value per
// provider and provision, `provision/<provider node>/<provision>`, with the provider's one user name; an offer with
// a login per consumer (ADR 0048, ADR 0158 §11, the default) is each consumer's own value,
// `pair/<provider node>/<provision>/<node>/<module>[/<local>]`, with the consumer's identity as its login (issue
// 677's notes 12 and 13). A credential the vault does not hold yet is waited for: nothing is made and no entry is
// written, so a consumer is never told of a login its provider has not made. A provider in the tool runner never asks
// the vault itself: its Vault is ConsumersFile, the file its machine's provisioner client fills (ADR 0373).
//
// **A multi-holder consumer gets no derived value** (ADR 0202 §10 as ADR 0349 clause 12 changes it): a consumer
// bound under a local name keeps several holders of the provision, each with its own login, while what the
// provider derives names one; its entry refuses it, naming both ends, and its client reports it.
//
// The helper speaks to the bus and the vault through the two interfaces below, so a provider wires it to its
// runtime and a test to a stand-in. Standard library only.
package provider
import (
"context"
"encoding/json"
"fmt"
"regexp"
"sort"
"strings"
"sync"
"time"
stdio "git.novox.be/novox/mesh-sdk/go"
)
// AssignmentsBucket is the controller's assignments mirror, `<node>.<module>`, whose value carries the plan of a
// module handed to the provisioner.
const AssignmentsBucket = "mesh-controller_assignments"
// BucketOf is a providing module's own bucket of entries, one per consumer: `<module>_bound`, the module's state
// `bound`.
func BucketOf(module string) string { return module + "_bound" }
// Bus is what the helper needs of the bus: a key-value watch and the provider's writes to its own bucket.
type Bus interface {
// Watch hands handle every current key and value of bucket, then each change as it comes, a removed key with a
// nil value; it returns once the current values are handed, and the watch lasts until ctx ends.
Watch(ctx context.Context, bucket string, handle func(key string, value []byte)) error
Put(ctx context.Context, bucket, key string, value []byte) error
Delete(ctx context.Context, bucket, key string) error
}
// Vault is how the provider reads a credential the vault keeps, by its name, opened.
type Vault interface {
Value(ctx context.Context, name string) ([]byte, error)
}
// CredentialForm is which credential an offer's consumers present.
type CredentialForm int
const (
// PerConsumer is a login of each consumer's own (ADR 0048, ADR 0158 §11): the default.
PerConsumer CredentialForm = iota
// Shared is the one credential the provider's software holds, shared with every consumer (ADR 0158).
Shared
// NoCredential is an offer whose consumers present none.
NoCredential
)
// Offer is what the provider serves one provision with.
type Offer struct {
// Serves are the values served to each consumer, the provider's config already composed into them; a string
// may name the consumer's identity, `${consumer:as}` or `${consumer:as:dns}`.
Serves map[string]any
// Credential is the form of the credential its consumers present.
Credential CredentialForm
// User is the provider's one user name, for a Shared credential.
User string
}
// Entry is one consumer's entry in the provider's bucket: the one statement both ends read. Never a secret.
type Entry struct {
Provider string `json:"provider"`
ProviderModule string `json:"provider-module"`
Provision string `json:"provision"`
Consumer string `json:"consumer"`
Module string `json:"module"`
Local string `json:"local,omitempty"`
// As is the consumer's identity, the mesh's (ADR 0202 §1).
As string `json:"as,omitempty"`
// User is the login the consumer presents: its identity, or the provider's one user name for a shared credential.
User string `json:"user,omitempty"`
// Values are the served values, composed for this consumer: what `${bound:<provision>:<key>}` reads.
Values map[string]any `json:"values,omitempty"`
// Credential is the name of the credential in the vault, which the consumer's client asks the vault for.
Credential string `json:"credential,omitempty"`
// Refused, when set, is why the consumer is not served, naming both ends.
Refused string `json:"refused,omitempty"`
// Contributes is what the consumer's module tells its provider about this provision (a database's name, its
// extensions), from its plan's entry (issue 776). Handed to Make and Gone; never written to the bucket.
Contributes map[string]any `json:"-"`
// At and From are where the consumer's plan says the provider is and which machine it is (issue 788): the values
// today's binding file gives that consumer. Written into Values as `at` and `from`; never alone to the bucket.
At, From string `json:"-"`
// Names are the names the controller composed for this consumer on this provision (issue 792): its own name, its
// internal name, and one of each per local name it contributes under, keyed as its binding file names them.
// Written into Values over anything the provider serves under the key; never alone to the bucket.
Names map[string]string `json:"-"`
}
// Key is the entry's key in the provider's bucket: `<node>.<module>.<provision>`, and `.<local>` for a bind under a
// local name.
func (e Entry) Key() string {
k := e.Consumer + "." + e.Module + "." + e.Provision
if e.Local != "" {
k += "." + e.Local
}
return k
}
// Consumer is what the provider's adapter makes: the consumer's entry and the credential the vault keeps for it.
type Consumer struct {
Entry Entry
Credential []byte
}
// Options are the provider's.
type Options struct {
// Node and Module are this provider: the machine it runs on and the module serving.
Node, Module string
// Offers are the provisions it serves, by name.
Offers map[string]Offer
Bus Bus
Vault Vault
// Make brings a consumer into being in the provider's software under the vault's credential. Asked on every
// pass for every consumer answered, so it must leave one that is already as asked alone; a rotated credential
// reaches the software this way.
Make func(ctx context.Context, c Consumer) error
// Gone, when set, hears that a consumer no longer binds the provider, before its entry is removed; an error
// keeps the entry, and it is heard again on the next pass. Retiring what was kept for it is the provider's
// (ADR 0230).
Gone func(ctx context.Context, e Entry) error
// Identity is the consumer's identity for an entry whose plan names none; ConsumerIdentity of the node and the
// module when not set.
Identity func(node, module string) string
// Every is how often a pass runs without a change watched; a minute when not set.
Every time.Duration
// Logf says what a pass did that differs from the last; the standard log when not set.
Logf func(format string, args ...any)
}
// Provider answers its consumers.
type Provider struct {
opts Options
mu sync.Mutex
assignments map[string][]byte
entries map[string][]byte
said map[string]string
poke chan struct{}
}
// New is a provider answering its consumers as opts say.
func New(opts Options) *Provider {
if opts.Identity == nil {
opts.Identity = ConsumerIdentity
}
if opts.Every <= 0 {
opts.Every = time.Minute
}
if opts.Logf == nil {
opts.Logf = func(f string, a ...any) { fmt.Printf("[provider] "+f+"\n", a...) }
}
return &Provider{opts: opts, assignments: map[string][]byte{}, entries: map[string][]byte{},
said: map[string]string{}, poke: make(chan struct{}, 1)}
}
// What a pass did for one bind.
const (
Answered = "answered"
Refused = "refused"
Waiting = "waiting"
Failed = "failed"
Removed = "removed"
)
// Outcome is what a pass did for one consumer's entry.
type Outcome struct {
Key string `json:"key"`
Status string `json:"status"`
Why string `json:"why,omitempty"`
}
// Run watches the binds and the provider's own entries, and passes once at start, on every change and every
// Every, until ctx ends.
func (p *Provider) Run(ctx context.Context) error {
if err := p.watch(ctx, BucketOf(p.opts.Module), p.entries); err != nil {
return fmt.Errorf("watching %s: %w", BucketOf(p.opts.Module), err)
}
if err := p.watch(ctx, AssignmentsBucket, p.assignments); err != nil {
return fmt.Errorf("watching %s: %w", AssignmentsBucket, err)
}
tick := time.NewTicker(p.opts.Every)
defer tick.Stop()
for {
p.mu.Lock()
assignments, entries := clone(p.assignments), clone(p.entries)
p.mu.Unlock()
if _, err := p.pass(ctx, assignments, entries); err != nil {
p.say("pass", err.Error())
}
select {
case <-ctx.Done():
return ctx.Err()
case <-p.poke:
case <-tick.C:
}
}
}
func (p *Provider) watch(ctx context.Context, bucket string, into map[string][]byte) error {
return p.opts.Bus.Watch(ctx, bucket, func(key string, value []byte) {
p.mu.Lock()
if value == nil {
delete(into, key)
} else {
into[key] = append([]byte(nil), value...)
}
p.mu.Unlock()
if bucket == AssignmentsBucket {
select {
case p.poke <- struct{}{}:
default:
}
}
})
}
// Pass reads the binds and the provider's entries from the bus and answers once: each bind of this provider gets
// its entry, and each entry of this provider no bind asks for is removed. An assignment that cannot be read stops
// the pass before anything is removed. Run passes by itself; Pass is for a provider that passes when it chooses.
func (p *Provider) Pass(ctx context.Context) ([]Outcome, error) {
watchCtx, cancel := context.WithCancel(ctx)
defer cancel()
var mu sync.Mutex
read := func(bucket string) (map[string][]byte, error) {
got := map[string][]byte{}
err := p.opts.Bus.Watch(watchCtx, bucket, func(key string, value []byte) {
mu.Lock()
defer mu.Unlock()
if value == nil {
delete(got, key)
} else {
got[key] = append([]byte(nil), value...)
}
})
mu.Lock()
defer mu.Unlock()
return clone(got), err
}
entries, err := read(BucketOf(p.opts.Module))
if err != nil {
return nil, fmt.Errorf("reading %s: %w", BucketOf(p.opts.Module), err)
}
assignments, err := read(AssignmentsBucket)
if err != nil {
return nil, fmt.Errorf("reading %s: %w", AssignmentsBucket, err)
}
return p.pass(ctx, assignments, entries)
}
// planned is as much of an assignment as the helper reads: lenient, since the plan is the client's to act on.
type planned struct {
Node string `json:"node"`
Module string `json:"module"`
Provisioning *struct {
Provisions []struct {
Provision string `json:"provision"`
Local string `json:"local"`
Provider string `json:"provider"`
ProviderModule string `json:"provider-module"`
Contributes map[string]any `json:"contributes"`
At string `json:"at"`
From string `json:"from"`
As string `json:"as"`
Names map[string]string `json:"names"`
} `json:"provisions"`
} `json:"provisioning"`
}
func (p *Provider) pass(ctx context.Context, assignments, entries map[string][]byte) ([]Outcome, error) {
var binds []Entry
for _, key := range sortedKeys(assignments) {
var a planned
if err := json.Unmarshal(assignments[key], &a); err != nil || a.Node+"."+a.Module != key {
return nil, fmt.Errorf("the assignment %s is not the controller's, so who binds %s on %s is not known; "+
"nothing is answered or removed", key, p.opts.Module, p.opts.Node)
}
if a.Provisioning == nil {
continue
}
for _, b := range a.Provisioning.Provisions {
if b.Provider != p.opts.Node || b.ProviderModule != p.opts.Module {
continue
}
binds = append(binds, Entry{Provider: p.opts.Node, ProviderModule: p.opts.Module,
Provision: b.Provision, Consumer: a.Node, Module: a.Module, Local: b.Local,
Contributes: b.Contributes, At: b.At, From: b.From, As: b.As, Names: b.Names})
}
}
var out []Outcome
want := map[string]bool{}
for _, b := range binds {
key := b.Key()
want[key] = true
o := p.answer(ctx, b, entries[key])
out = append(out, o)
p.say(key, o.Status+" "+o.Why)
}
for _, key := range sortedKeys(entries) {
if want[key] {
continue
}
var e Entry
if err := json.Unmarshal(entries[key], &e); err != nil || e.Provider != p.opts.Node ||
e.ProviderModule != p.opts.Module {
continue // another machine's instance of this provider wrote it, or nothing of this helper's
}
if p.opts.Gone != nil {
if err := p.opts.Gone(ctx, e); err != nil {
o := Outcome{Key: key, Status: Failed, Why: "no longer bound, and the provider could not let it go: " +
err.Error()}
out = append(out, o)
p.say(key, o.Status+" "+o.Why)
continue
}
}
if err := p.opts.Bus.Delete(ctx, BucketOf(p.opts.Module), key); err != nil {
o := Outcome{Key: key, Status: Failed, Why: "no longer bound, and its entry could not be removed: " +
err.Error()}
out = append(out, o)
p.say(key, o.Status+" "+o.Why)
continue
}
out = append(out, Outcome{Key: key, Status: Removed})
p.say(key, Removed)
p.mu.Lock()
delete(p.said, key)
p.mu.Unlock()
}
return out, nil
}
// answer makes one bind's entry and writes it when it differs from what the bucket holds.
func (p *Provider) answer(ctx context.Context, e Entry, have []byte) Outcome {
key := e.Key()
offer, offered := p.opts.Offers[e.Provision]
switch {
case !offered:
e.Refused = fmt.Sprintf("%s on %s is bound to %s on %s for %s, which %s does not serve", e.Module,
e.Consumer, e.ProviderModule, e.Provider, e.Provision, e.ProviderModule)
case e.Local != "" && derives(offer.Serves):
e.Refused = fmt.Sprintf("%s on %s keeps several holders of %s (this one is %q), and %s on %s derives %s for "+
"each consumer from its login: each holder has a login of its own and a consumer is told one value per "+
"requirement, so the two ends would name different things (novox/hq ADR 0202 §10, ADR 0349 clause 12)",
e.Module, e.Consumer, e.Provision, e.Local, e.ProviderModule, e.Provider,
strings.Join(derivedKeys(offer.Serves), ", "))
}
if e.Refused != "" {
return p.write(ctx, e, have, Outcome{Key: key, Status: Refused, Why: e.Refused})
}
// The plan's identity is the controller's, slug first; Identity only answers for an entry without one.
if e.As == "" {
e.As = p.opts.Identity(e.Consumer, e.Module)
}
values, err := ServedTo(offer.Serves, e.As)
if err != nil {
e.Refused = fmt.Sprintf("%s on %s serving %s to %s on %s: %v", e.ProviderModule, e.Provider, e.Provision,
e.Module, e.Consumer, err)
e.As = ""
return p.write(ctx, e, have, Outcome{Key: key, Status: Refused, Why: e.Refused})
}
if e.At != "" || e.From != "" {
if values == nil {
values = map[string]any{}
}
if e.At != "" {
values["at"] = e.At
}
if e.From != "" {
values["from"] = e.From
}
}
if len(e.Names) > 0 {
if values == nil {
values = map[string]any{}
}
for key, name := range e.Names {
values[key] = name
}
}
e.Values = values
var credential []byte
switch offer.Credential {
case Shared:
e.User = offer.User
e.Credential = SharedName(e.Provider, e.Provision)
case PerConsumer:
e.User = e.As
e.Credential = PairName(e.Provider, e.Provision, e.Consumer, e.Module, e.Local)
}
if e.Credential != "" {
credential, err = p.opts.Vault.Value(ctx, e.Credential)
if err != nil {
return Outcome{Key: key, Status: Waiting, Why: fmt.Sprintf("the vault's %s: %v", e.Credential, err)}
}
}
if p.opts.Make != nil {
if err := p.opts.Make(ctx, Consumer{Entry: e, Credential: credential}); err != nil {
why := err.Error()
if len(credential) > 0 {
why = strings.ReplaceAll(why, string(credential), "<credential>")
}
return Outcome{Key: key, Status: Failed, Why: "the provider could not make it: " + why}
}
}
return p.write(ctx, e, have, Outcome{Key: key, Status: Answered})
}
func (p *Provider) write(ctx context.Context, e Entry, have []byte, o Outcome) Outcome {
body, err := json.Marshal(e)
if err != nil {
return Outcome{Key: o.Key, Status: Failed, Why: err.Error()}
}
if sameJSON(have, body) {
return o
}
if err := p.opts.Bus.Put(ctx, BucketOf(p.opts.Module), o.Key, body); err != nil {
return Outcome{Key: o.Key, Status: Failed, Why: "its entry could not be written: " + err.Error()}
}
return o
}
// say logs what a key's pass said when it differs from the last.
func (p *Provider) say(key, what string) {
p.mu.Lock()
same := p.said[key] == what
p.said[key] = what
p.mu.Unlock()
if !same {
p.opts.Logf("%s: %s", key, strings.TrimSpace(what))
}
}
// SharedName is the vault's one value per provider and provision (ADR 0158, issue 677).
func SharedName(provider, provision string) string { return "provision/" + provider + "/" + provision }
// PairName is the vault's value of one consumer's own login (ADR 0048, ADR 0158 §11, issue 677's note 12).
func PairName(provider, provision, node, module, local string) string {
n := "pair/" + provider + "/" + provision + "/" + node + "/" + module
if local != "" {
n += "/" + local
}
return n
}
// identityUnusable is every character a consumer's identity does not hold (mesh-controller internal/catalogue
// identity.go, the one derivation).
var identityUnusable = regexp.MustCompile(`[^a-z0-9_]+`)
// ConsumerIdentity is what one module on one machine is called wherever it authenticates: `mesh_<node>_<module>`,
// lower case, anything but letters, digits and `_` written `_` (mesh-controller ConsumerIdentity). module is the
// identity source: the module's slug when it declares one, otherwise its name (ADR 0049).
func ConsumerIdentity(node, module string) string {
clean := func(s string) string {
return strings.Trim(identityUnusable.ReplaceAllString(strings.ToLower(s), "_"), "_")
}
return "mesh_" + clean(node) + "_" + clean(module)
}
// consumerFact is `${consumer:<fact>}` or `${consumer:<fact>:<alphabet>}`.
var consumerFact = regexp.MustCompile(`\$\{consumer:([a-z][a-z0-9-]*)(?::([a-z][a-z0-9-]*))?\}`)
// ServedTo fills an offer's served values for one consumer whose identity is as (ADR 0202 §1): `${consumer:as}` as
// it is, `${consumer:as:dns}` with `-` for `_`. Only strings carry placeholders; any other fact or alphabet is
// refused, never left standing.
func ServedTo(serves map[string]any, as string) (map[string]any, error) {
if serves == nil {
return nil, nil
}
out := make(map[string]any, len(serves))
for _, key := range sortedKeys(serves) {
text, ok := serves[key].(string)
if !ok || !strings.Contains(text, "${consumer:") {
out[key] = serves[key]
continue
}
var failed error
out[key] = consumerFact.ReplaceAllStringFunc(text, func(match string) string {
parts := consumerFact.FindStringSubmatch(match)
switch {
case parts[1] == "as" && parts[2] == "":
return as
case parts[1] == "as" && parts[2] == "dns":
return strings.ReplaceAll(as, "_", "-")
}
if failed == nil {
failed = fmt.Errorf("the value served as %q says %s, and the mesh states only ${consumer:as} and "+
"${consumer:as:dns}", key, match)
}
return match
})
if failed != nil {
return nil, failed
}
}
return out, nil
}
func derives(serves map[string]any) bool { return len(derivedKeys(serves)) > 0 }
func derivedKeys(serves map[string]any) []string {
var out []string
for _, k := range sortedKeys(serves) {
if text, ok := serves[k].(string); ok && strings.Contains(text, "${consumer:") {
out = append(out, k)
}
}
return out
}
func clone(m map[string][]byte) map[string][]byte {
out := make(map[string][]byte, len(m))
for k, v := range m {
out[k] = v
}
return out
}
func sameJSON(a, b []byte) bool {
if a == nil || b == nil {
return false
}
var x, y any
if json.Unmarshal(a, &x) != nil || json.Unmarshal(b, &y) != nil {
return false
}
ja, _ := json.Marshal(x)
jb, _ := json.Marshal(y)
return string(ja) == string(jb)
}
func sortedKeys[V any](m map[string]V) []string {
if len(m) == 0 {
return nil
}
out := make([]string, 0, len(m))
for k := range m {
out = append(out, k)
}
sort.Strings(out)
return out
}
// StateBus is the Bus of a provider that runs as a module bundle over the runtime's stdio: the provider's own bucket,
// BucketOf(module), is the module's state `bound`, and AssignmentsBucket is the state it reads,
// `mesh-controller.assignments` (both declared in the module's manifest). Any other bucket is refused.
//
// A stdio state watch lasts as long as the process and cannot be ended (stdio.KeptState.Watch), so Watch honours the
// Bus contract by silencing instead: once ctx has ended the handler is never called again, though the runtime keeps
// sending changes that are dropped. A Pass, which watches twice per call, is therefore correct, only the dropped
// changes of its two spent watches remain.
func StateBus(module string) Bus { return stateBus{module: module} }
type stateBus struct{ module string }
func (b stateBus) state(bucket string) (stdio.KeptState, error) {
switch bucket {
case BucketOf(b.module):
return stdio.State("bound"), nil
case AssignmentsBucket:
return stdio.State("mesh-controller.assignments"), nil
}
return stdio.KeptState{}, fmt.Errorf("%s is not a bucket this provider keeps or reads", bucket)
}
func (b stateBus) Watch(ctx context.Context, bucket string, handle func(key string, value []byte)) error {
s, err := b.state(bucket)
if err != nil {
return err
}
return s.Watch("", func(ch stdio.StateChange) error {
if ctx.Err() != nil {
return nil
}
if ch.Op == "delete" {
handle(ch.Key, nil)
} else {
handle(ch.Key, ch.Value)
}
return nil
})
}
func (b stateBus) Put(_ context.Context, bucket, key string, value []byte) error {
s, err := b.state(bucket)
if err != nil {
return err
}
_, err = s.Put(key, json.RawMessage(value))
return err
}
func (b stateBus) Delete(_ context.Context, bucket, key string) error {
s, err := b.state(bucket)
if err != nil {
return err
}
return s.Delete(key)
}
+590
View File
@@ -0,0 +1,590 @@
package provider
import (
"bytes"
"context"
"encoding/json"
"errors"
"fmt"
"reflect"
"slices"
"strings"
"sync"
"testing"
"time"
)
// standIn is a stand-in bus: buckets of keys and values, and watches that hand the current values first and then
// every change, as the bus's key-value watch does.
type standIn struct {
mu sync.Mutex
buckets map[string]map[string][]byte
watchers map[string][]func(string, []byte)
}
func newStandIn() *standIn {
return &standIn{buckets: map[string]map[string][]byte{}, watchers: map[string][]func(string, []byte){}}
}
func (s *standIn) Watch(ctx context.Context, bucket string, handle func(key string, value []byte)) error {
s.mu.Lock()
current := map[string][]byte{}
for k, v := range s.buckets[bucket] {
current[k] = v
}
s.watchers[bucket] = append(s.watchers[bucket], handle)
s.mu.Unlock()
for _, k := range sortedKeys(current) {
handle(k, current[k])
}
return nil
}
func (s *standIn) Put(_ context.Context, bucket, key string, value []byte) error {
s.mu.Lock()
if s.buckets[bucket] == nil {
s.buckets[bucket] = map[string][]byte{}
}
s.buckets[bucket][key] = append([]byte(nil), value...)
ws := slices.Clone(s.watchers[bucket])
s.mu.Unlock()
for _, w := range ws {
w(key, value)
}
return nil
}
func (s *standIn) Delete(_ context.Context, bucket, key string) error {
s.mu.Lock()
delete(s.buckets[bucket], key)
ws := slices.Clone(s.watchers[bucket])
s.mu.Unlock()
for _, w := range ws {
w(key, nil)
}
return nil
}
func (s *standIn) get(bucket, key string) []byte {
s.mu.Lock()
defer s.mu.Unlock()
return s.buckets[bucket][key]
}
func (s *standIn) keys(bucket string) []string {
s.mu.Lock()
defer s.mu.Unlock()
return sortedKeys(s.buckets[bucket])
}
// everything is every value the bus holds, for a test that no secret is in a bucket.
func (s *standIn) everything() []byte {
s.mu.Lock()
defer s.mu.Unlock()
var all []byte
for _, b := range s.buckets {
for k, v := range b {
all = append(all, k...)
all = append(all, v...)
}
}
return all
}
// vault is a stand-in vault: a value per name, or nothing yet.
type vault struct {
mu sync.Mutex
values map[string]string
asked []string
}
func (v *vault) Value(_ context.Context, name string) ([]byte, error) {
v.mu.Lock()
defer v.mu.Unlock()
v.asked = append(v.asked, name)
val, ok := v.values[name]
if !ok {
return nil, fmt.Errorf("the vault holds no %s yet", name)
}
return []byte(val), nil
}
// maker records what the provider's adapter was asked to make, with the credential it was handed.
type maker struct {
mu sync.Mutex
made map[string]string
// contributed is what each consumer's entry carried of its plan's contribution.
contributed map[string]map[string]any
// as is the identity each consumer's entry was made under.
as map[string]string
gone []string
fail error
calls int
}
func (m *maker) Make(_ context.Context, c Consumer) error {
m.mu.Lock()
defer m.mu.Unlock()
m.calls++
if m.fail != nil {
return m.fail
}
if m.made == nil {
m.made = map[string]string{}
}
m.made[c.Entry.Key()] = string(c.Credential)
if m.contributed == nil {
m.contributed = map[string]map[string]any{}
}
m.contributed[c.Entry.Key()] = c.Entry.Contributes
if m.as == nil {
m.as = map[string]string{}
}
m.as[c.Entry.Key()] = c.Entry.As
return nil
}
func (m *maker) Gone(_ context.Context, e Entry) error {
m.mu.Lock()
defer m.mu.Unlock()
m.gone = append(m.gone, e.Key())
return nil
}
// assign puts a consumer's assignment with its plan's provisions into the stand-in mirror.
func assign(t *testing.T, bus *standIn, node, module string, provisions ...map[string]string) {
t.Helper()
ps := []map[string]string{}
ps = append(ps, provisions...)
raw, err := json.Marshal(map[string]any{"node": node, "module": module, "assigned": "2026-10-15T00:00:00Z",
"provisioning": map[string]any{"templates": map[string]string{"source": "x", "digest": "y"},
"provisions": ps}})
if err != nil {
t.Fatal(err)
}
if err := bus.Put(context.Background(), AssignmentsBucket, node+"."+module, raw); err != nil {
t.Fatal(err)
}
}
func bind(provision, local, provider, providerModule string) map[string]string {
b := map[string]string{"provision": provision, "provider": provider, "provider-module": providerModule}
if local != "" {
b["local"] = local
}
return b
}
func entryOf(t *testing.T, bus *standIn, key string) Entry {
t.Helper()
raw := bus.get(BucketOf("postgres"), key)
if raw == nil {
t.Fatalf("no entry %s in %s; it holds %v", key, BucketOf("postgres"), bus.keys(BucketOf("postgres")))
}
var e Entry
if err := json.Unmarshal(raw, &e); err != nil {
t.Fatalf("entry %s: %v", key, err)
}
return e
}
// postgres is a provider of postgres-database on novox with a login per consumer (ADR 0048, ADR 0158 §11), its
// served values derived from each consumer's identity (ADR 0202 §1).
func postgres(bus *standIn, v *vault, m *maker) *Provider {
return New(Options{
Node: "novox", Module: "postgres", Bus: bus, Vault: v, Make: m.Make, Gone: m.Gone,
Offers: map[string]Offer{
"postgres-database": {Credential: PerConsumer, Serves: map[string]any{
"host": "postgres.novox.mesh", "port": 5432, "database": "${consumer:as}"}},
},
Logf: func(string, ...any) {},
})
}
// A provider answers each consumer that binds it with an entry of its own in its own bucket: composed from its
// served values and the consumer's identity, naming the credential the vault keeps for that consumer, and never
// the credential itself; its adapter is handed each consumer's own value read from the vault.
func TestEachConsumerGetsAnEntryOfItsOwnAndTheVaultsValue(t *testing.T) {
bus, m := newStandIn(), &maker{}
v := &vault{values: map[string]string{
"pair/novox/postgres-database/ace/letta": "letta-secret-1",
"pair/novox/postgres-database/g14/umami": "umami-secret-2",
}}
assign(t, bus, "ace", "letta", bind("postgres-database", "", "novox", "postgres"))
assign(t, bus, "g14", "umami", bind("postgres-database", "", "novox", "postgres"))
p := postgres(bus, v, m)
out, err := p.Pass(context.Background())
if err != nil {
t.Fatal(err)
}
if got := bus.keys(BucketOf("postgres")); !slices.Equal(got, []string{"ace.letta.postgres-database",
"g14.umami.postgres-database"}) {
t.Fatalf("entries %v", got)
}
e := entryOf(t, bus, "ace.letta.postgres-database")
if e.Provider != "novox" || e.ProviderModule != "postgres" || e.Consumer != "ace" || e.Module != "letta" ||
e.As != "mesh_ace_letta" || e.User != "mesh_ace_letta" ||
e.Credential != "pair/novox/postgres-database/ace/letta" || e.Refused != "" {
t.Fatalf("entry %+v", e)
}
if e.Values["database"] != "mesh_ace_letta" || e.Values["host"] != "postgres.novox.mesh" ||
e.Values["port"] != float64(5432) {
t.Fatalf("values %v", e.Values)
}
if m.made["ace.letta.postgres-database"] != "letta-secret-1" ||
m.made["g14.umami.postgres-database"] != "umami-secret-2" {
t.Fatalf("made %v", m.made)
}
if bytes.Contains(bus.everything(), []byte("secret-")) {
t.Fatal("a credential is in a bucket")
}
for _, o := range out {
if o.Status != Answered {
t.Fatalf("outcome %+v", o)
}
}
}
// A provider whose offer shares one credential (ADR 0158) names the vault's one value for the provision, and its
// one user name, to every consumer.
func TestASharedCredentialIsTheVaultsOneValueAndTheProvidersOneUser(t *testing.T) {
bus, m := newStandIn(), &maker{}
v := &vault{values: map[string]string{"provision/novox/sonarr-api": "one-value"}}
assign(t, bus, "ace", "overseerr", bind("sonarr-api", "", "novox", "sonarr"))
assign(t, bus, "g14", "homepage", bind("sonarr-api", "", "novox", "sonarr"))
p := New(Options{Node: "novox", Module: "sonarr", Bus: bus, Vault: v, Make: m.Make, Logf: func(string, ...any) {},
Offers: map[string]Offer{"sonarr-api": {Credential: Shared, User: "sonarr",
Serves: map[string]any{"url": "http://sonarr.novox.mesh:8989"}}}})
if _, err := p.Pass(context.Background()); err != nil {
t.Fatal(err)
}
for _, key := range []string{"ace.overseerr.sonarr-api", "g14.homepage.sonarr-api"} {
var e Entry
if err := json.Unmarshal(bus.get(BucketOf("sonarr"), key), &e); err != nil {
t.Fatalf("%s: %v", key, err)
}
if e.Credential != "provision/novox/sonarr-api" || e.User != "sonarr" || e.Values["url"] == nil {
t.Fatalf("%s: %+v", key, e)
}
if m.made[key] != "one-value" {
t.Fatalf("%s made with %q", key, m.made[key])
}
}
if bytes.Contains(bus.everything(), []byte("one-value")) {
t.Fatal("the shared credential is in a bucket")
}
}
// A consumer that keeps several holders of one provision is served no derived value: its entry refuses it, naming
// both ends, so its client can report it (ADR 0349 clause 12, ADR 0202 §10); nothing is made for it.
func TestAMultiHolderConsumerIsRefusedADerivedValueNamingBothEnds(t *testing.T) {
bus, m := newStandIn(), &maker{}
v := &vault{values: map[string]string{"pair/novox/postgres-database/ace/letta/archive": "x-secret"}}
assign(t, bus, "ace", "letta", bind("postgres-database", "archive", "novox", "postgres"))
out, err := postgres(bus, v, m).Pass(context.Background())
if err != nil {
t.Fatal(err)
}
e := entryOf(t, bus, "ace.letta.postgres-database.archive")
for _, says := range []string{"letta on ace", "postgres on novox", "postgres-database", "archive"} {
if !strings.Contains(e.Refused, says) {
t.Fatalf("the refusal does not name %q: %s", says, e.Refused)
}
}
if e.Values != nil || e.Credential != "" {
t.Fatalf("a refused entry carries values: %+v", e)
}
if m.calls != 0 {
t.Fatal("the adapter made a refused consumer")
}
if len(out) != 1 || out[0].Status != Refused {
t.Fatalf("outcomes %+v", out)
}
}
// A consumer bound under a local name is served when the offer derives nothing from it: its credential is its own
// pair row under that local name.
func TestALocalNameIsServedWhenNothingIsDerived(t *testing.T) {
bus, m := newStandIn(), &maker{}
v := &vault{values: map[string]string{"pair/novox/redis-cache/ace/letta/queue": "q-secret"}}
assign(t, bus, "ace", "letta", bind("redis-cache", "queue", "novox", "redis"))
p := New(Options{Node: "novox", Module: "redis", Bus: bus, Vault: v, Make: m.Make, Logf: func(string, ...any) {},
Offers: map[string]Offer{"redis-cache": {Credential: PerConsumer, Serves: map[string]any{"port": 6379}}}})
if _, err := p.Pass(context.Background()); err != nil {
t.Fatal(err)
}
var e Entry
if err := json.Unmarshal(bus.get(BucketOf("redis"), "ace.letta.redis-cache.queue"), &e); err != nil {
t.Fatal(err)
}
if e.Refused != "" || e.Credential != "pair/novox/redis-cache/ace/letta/queue" || e.Local != "queue" {
t.Fatalf("entry %+v", e)
}
if m.made["ace.letta.redis-cache.queue"] != "q-secret" {
t.Fatalf("made %v", m.made)
}
}
// A consumer that unbinds loses its entry and the adapter hears it is gone; an entry another machine's instance of
// the same provider wrote is not this one's to remove.
func TestAnUnbindRemovesTheEntryAndLeavesAnotherProvidersAlone(t *testing.T) {
bus, m := newStandIn(), &maker{}
v := &vault{values: map[string]string{"pair/novox/postgres-database/ace/letta": "s1"}}
assign(t, bus, "ace", "letta", bind("postgres-database", "", "novox", "postgres"))
other, _ := json.Marshal(Entry{Provider: "shanks", ProviderModule: "postgres", Provision: "postgres-database",
Consumer: "g14", Module: "gitea"})
_ = bus.Put(context.Background(), BucketOf("postgres"), "g14.gitea.postgres-database", other)
p := postgres(bus, v, m)
if _, err := p.Pass(context.Background()); err != nil {
t.Fatal(err)
}
_ = bus.Delete(context.Background(), AssignmentsBucket, "ace.letta")
out, err := p.Pass(context.Background())
if err != nil {
t.Fatal(err)
}
if got := bus.keys(BucketOf("postgres")); !slices.Equal(got, []string{"g14.gitea.postgres-database"}) {
t.Fatalf("entries left %v", got)
}
if !slices.Equal(m.gone, []string{"ace.letta.postgres-database"}) {
t.Fatalf("gone %v", m.gone)
}
if len(out) != 1 || out[0].Status != Removed {
t.Fatalf("outcomes %+v", out)
}
}
// A bind whose credential the vault does not hold yet is waited for: nothing is made and no entry is written, so a
// consumer is never told of a login its provider has not made; once the vault holds it, it is answered.
func TestACredentialTheVaultDoesNotHoldYetIsWaitedFor(t *testing.T) {
bus, m := newStandIn(), &maker{}
v := &vault{values: map[string]string{}}
assign(t, bus, "ace", "letta", bind("postgres-database", "", "novox", "postgres"))
p := postgres(bus, v, m)
out, err := p.Pass(context.Background())
if err != nil {
t.Fatal(err)
}
if len(out) != 1 || out[0].Status != Waiting || !strings.Contains(out[0].Why, "pair/novox/postgres-database/ace/letta") {
t.Fatalf("outcomes %+v", out)
}
if bus.keys(BucketOf("postgres")) != nil || m.calls != 0 {
t.Fatal("a consumer was answered before its credential was there")
}
v.values["pair/novox/postgres-database/ace/letta"] = "late"
if _, err := p.Pass(context.Background()); err != nil {
t.Fatal(err)
}
if entryOf(t, bus, "ace.letta.postgres-database").Refused != "" || m.made["ace.letta.postgres-database"] != "late" {
t.Fatal("not answered once the vault held the value")
}
}
// Only the binds naming this provider, on this machine, are its own; a bind naming it for a provision it does not
// offer is refused, naming both ends.
func TestOnlyBindsNamingThisProviderAreAnswered(t *testing.T) {
bus, m := newStandIn(), &maker{}
v := &vault{values: map[string]string{"pair/novox/postgres-database/ace/letta": "s"}}
assign(t, bus, "ace", "letta", bind("postgres-database", "", "shanks", "postgres"),
bind("mongodb-database", "", "novox", "postgres"))
assign(t, bus, "ace", "other", bind("postgres-database", "", "novox", "mssql"))
out, err := postgres(bus, v, m).Pass(context.Background())
if err != nil {
t.Fatal(err)
}
if got := bus.keys(BucketOf("postgres")); !slices.Equal(got, []string{"ace.letta.mongodb-database"}) {
t.Fatalf("entries %v", got)
}
e := entryOf(t, bus, "ace.letta.mongodb-database")
if !strings.Contains(e.Refused, "postgres on novox") || !strings.Contains(e.Refused, "letta on ace") {
t.Fatalf("refusal %q", e.Refused)
}
if len(out) != 1 || m.calls != 0 {
t.Fatalf("outcomes %+v, %d made", out, m.calls)
}
}
// An assignment that cannot be read stops the pass: a bind missed would read as an unbind and remove an entry its
// consumer still reads.
func TestAnUnreadableAssignmentRemovesNothing(t *testing.T) {
bus, m := newStandIn(), &maker{}
v := &vault{values: map[string]string{"pair/novox/postgres-database/ace/letta": "s"}}
assign(t, bus, "ace", "letta", bind("postgres-database", "", "novox", "postgres"))
p := postgres(bus, v, m)
if _, err := p.Pass(context.Background()); err != nil {
t.Fatal(err)
}
_ = bus.Put(context.Background(), AssignmentsBucket, "ace.letta", []byte("{not json"))
if _, err := p.Pass(context.Background()); err == nil {
t.Fatal("a pass over an unreadable assignment went on")
}
if bus.get(BucketOf("postgres"), "ace.letta.postgres-database") == nil || len(m.gone) != 0 {
t.Fatal("an entry was removed on an unreadable assignment")
}
}
// An adapter that fails is said with the credential taken out of its words, and the entry is not written.
func TestAFailedMakeIsSaidWithoutTheCredential(t *testing.T) {
bus, m := newStandIn(), &maker{fail: errors.New(`password "hunter2-secret" rejected`)}
v := &vault{values: map[string]string{"pair/novox/postgres-database/ace/letta": "hunter2-secret"}}
assign(t, bus, "ace", "letta", bind("postgres-database", "", "novox", "postgres"))
var logged []string
p := postgres(bus, v, m)
p.opts.Logf = func(f string, a ...any) { logged = append(logged, fmt.Sprintf(f, a...)) }
out, err := p.Pass(context.Background())
if err != nil {
t.Fatal(err)
}
if len(out) != 1 || out[0].Status != Failed || strings.Contains(out[0].Why, "hunter2") {
t.Fatalf("outcomes %+v", out)
}
if strings.Contains(strings.Join(logged, "\n"), "hunter2") || len(logged) == 0 {
t.Fatalf("logged %q", logged)
}
if bus.keys(BucketOf("postgres")) != nil {
t.Fatal("an entry was written for a consumer the adapter could not make")
}
}
// Run watches the binds on the stand-in bus: a consumer that binds is answered, and one that unbinds is removed,
// without a pass asked by hand.
func TestRunAnswersBindsAndUnbindsAsTheyAreWatched(t *testing.T) {
bus, m := newStandIn(), &maker{}
v := &vault{values: map[string]string{"pair/novox/postgres-database/ace/letta": "s"}}
p := postgres(bus, v, m)
p.opts.Every = time.Hour
ctx, cancel := context.WithCancel(context.Background())
done := make(chan error, 1)
go func() { done <- p.Run(ctx) }()
assign(t, bus, "ace", "letta", bind("postgres-database", "", "novox", "postgres"))
waitFor(t, func() bool { return bus.get(BucketOf("postgres"), "ace.letta.postgres-database") != nil })
_ = bus.Delete(context.Background(), AssignmentsBucket, "ace.letta")
waitFor(t, func() bool { return bus.get(BucketOf("postgres"), "ace.letta.postgres-database") == nil })
cancel()
if err := <-done; err != nil && !errors.Is(err, context.Canceled) {
t.Fatal(err)
}
}
func waitFor(t *testing.T, ok func() bool) {
t.Helper()
deadline := time.Now().Add(3 * time.Second)
for !ok() {
if time.Now().After(deadline) {
t.Fatal("not within 3s")
}
time.Sleep(5 * time.Millisecond)
}
}
// The consumer's identity is the mesh's (ADR 0202 §1): mesh_<node>_<module>, lower case, `_` for anything else,
// and its DNS form writes `-` for `_`.
func TestTheIdentityIsTheMeshsAndItsDNSForm(t *testing.T) {
if got := ConsumerIdentity("Home.Server", "my-app"); got != "mesh_home_server_my_app" {
t.Fatal(got)
}
got, err := ServedTo(map[string]any{"bucket": "b-${consumer:as:dns}", "n": 3}, "mesh_ace_letta")
if err != nil || got["bucket"] != "b-mesh-ace-letta" || got["n"] != 3 {
t.Fatalf("%v %v", got, err)
}
if _, err := ServedTo(map[string]any{"x": "${consumer:node}"}, "mesh_ace_letta"); err == nil {
t.Fatal("a fact the mesh does not state was filled")
}
}
// A consumer's plan entry may carry what its module contributes to the provider (issue 776): the helper hands it to
// the provider's adapter with the consumer, and does not write it to the bucket.
func TestAConsumersContributionIsHandedToTheProviderPerConsumer(t *testing.T) {
bus, m := newStandIn(), &maker{}
v := &vault{values: map[string]string{
"pair/novox/postgres-database/ace/nextcloud": "a", "pair/novox/postgres-database/g14/umami": "b"}}
raw := `{"node":"ace","module":"nextcloud","provisioning":{"templates":{"source":"x","digest":"y"},` +
`"provisions":[{"provision":"postgres-database","provider":"novox","provider-module":"postgres",` +
`"contributes":{"name":"nextcloud","extensions":["pg_trgm"]}}]}}`
if err := bus.Put(context.Background(), AssignmentsBucket, "ace.nextcloud", []byte(raw)); err != nil {
t.Fatal(err)
}
assign(t, bus, "g14", "umami", bind("postgres-database", "", "novox", "postgres"))
if _, err := postgres(bus, v, m).Pass(context.Background()); err != nil {
t.Fatal(err)
}
want := map[string]any{"name": "nextcloud", "extensions": []any{"pg_trgm"}}
if got := m.contributed["ace.nextcloud.postgres-database"]; !reflect.DeepEqual(got, want) {
t.Fatalf("the adapter was handed the contribution %v, want %v", got, want)
}
if got := m.contributed["g14.umami.postgres-database"]; got != nil {
t.Fatalf("a consumer contributing nothing was handed %v", got)
}
if bytes.Contains(bus.get(BucketOf("postgres"), "ace.nextcloud.postgres-database"), []byte("pg_trgm")) {
t.Fatal("a contribution is in the provider's bucket")
}
}
// A consumer's plan entry carries where its provider is, which machine, and who the consumer is (issue 788): the
// entry's values carry `at` and `from` as the binding file gives them, and its `as` is the plan's, so a module with a
// slug keeps the login it has today. An entry of a plan without them is as it was.
func TestAConsumersEntryCarriesAtFromAndTheAsOfItsPlan(t *testing.T) {
bus, m := newStandIn(), &maker{}
v := &vault{values: map[string]string{
"pair/novox/postgres-database/ace/claude-code": "a", "pair/novox/postgres-database/g14/umami": "b"}}
raw := `{"node":"ace","module":"claude-code","provisioning":{"templates":{"source":"x","digest":"y"},` +
`"provisions":[{"provision":"postgres-database","provider":"novox","provider-module":"postgres",` +
`"at":"10.7.0.1","from":"novox","as":"mesh_ace_agent"}]}}`
if err := bus.Put(context.Background(), AssignmentsBucket, "ace.claude-code", []byte(raw)); err != nil {
t.Fatal(err)
}
assign(t, bus, "g14", "umami", bind("postgres-database", "", "novox", "postgres"))
if _, err := postgres(bus, v, m).Pass(context.Background()); err != nil {
t.Fatal(err)
}
var e Entry
if err := json.Unmarshal(bus.get(BucketOf("postgres"), "ace.claude-code.postgres-database"), &e); err != nil {
t.Fatal(err)
}
if e.Values["at"] != "10.7.0.1" || e.Values["from"] != "novox" || e.Values["port"] == nil {
t.Fatalf("the entry's values are %v", e.Values)
}
if e.As != "mesh_ace_agent" || e.User != "mesh_ace_agent" || m.as["ace.claude-code.postgres-database"] != "mesh_ace_agent" {
t.Fatalf("the entry is %+v, made as %q", e, m.as["ace.claude-code.postgres-database"])
}
var plain Entry
if err := json.Unmarshal(bus.get(BucketOf("postgres"), "g14.umami.postgres-database"), &plain); err != nil {
t.Fatal(err)
}
if _, has := plain.Values["at"]; has || plain.As != "mesh_g14_umami" {
t.Fatalf("an entry of a plan without them changed: %+v", plain)
}
}
// A consumer's plan entry may carry the names the controller composed for it (issue 792): each is copied into the
// entry's values under its key, over anything the provider serves under it. An entry without them is as it was.
func TestAConsumersEntryCarriesTheNamesOfItsPlan(t *testing.T) {
bus, m := newStandIn(), &maker{}
v := &vault{values: map[string]string{
"pair/novox/postgres-database/ace/gitea": "a", "pair/novox/postgres-database/g14/umami": "b"}}
raw := `{"node":"ace","module":"gitea","provisioning":{"templates":{"source":"x","digest":"y"},` +
`"provisions":[{"provision":"postgres-database","provider":"novox","provider-module":"postgres",` +
`"names":{"name":"git.example.org","internal-name":"git.mesh","name-web":"web.example.org"}}]}}`
if err := bus.Put(context.Background(), AssignmentsBucket, "ace.gitea", []byte(raw)); err != nil {
t.Fatal(err)
}
assign(t, bus, "g14", "umami", bind("postgres-database", "", "novox", "postgres"))
if _, err := postgres(bus, v, m).Pass(context.Background()); err != nil {
t.Fatal(err)
}
var e Entry
if err := json.Unmarshal(bus.get(BucketOf("postgres"), "ace.gitea.postgres-database"), &e); err != nil {
t.Fatal(err)
}
if e.Values["name"] != "git.example.org" || e.Values["internal-name"] != "git.mesh" ||
e.Values["name-web"] != "web.example.org" || e.Values["port"] == nil {
t.Fatalf("the entry's values are %v", e.Values)
}
var plain Entry
if err := json.Unmarshal(bus.get(BucketOf("postgres"), "g14.umami.postgres-database"), &plain); err != nil {
t.Fatal(err)
}
if _, has := plain.Values["name"]; has {
t.Fatalf("an entry of a plan without names changed: %+v", plain)
}
}
+152
View File
@@ -0,0 +1,152 @@
package provider
import (
"bufio"
"context"
"encoding/json"
"io"
"strings"
"testing"
"time"
stdio "git.novox.be/novox/mesh-sdk/go"
)
// runtime drives a bundle over pipes as the runtime does: it reads what the bundle says and sends what the runtime would.
type runtime struct {
t *testing.T
lines *bufio.Scanner
in io.Writer
}
func (r *runtime) next() map[string]any {
done := make(chan map[string]any, 1)
go func() {
if r.lines.Scan() {
var m map[string]any
_ = json.Unmarshal(r.lines.Bytes(), &m)
done <- m
}
}()
select {
case m := <-done:
return m
case <-time.After(3 * time.Second):
r.t.Fatal("the bundle said nothing")
return nil
}
}
func (r *runtime) send(m map[string]any) {
b, _ := json.Marshal(m)
_, _ = r.in.Write(append(b, '\n'))
}
func (r *runtime) call(id int, tool string, args map[string]any) {
r.send(map[string]any{"jsonrpc": "2.0", "id": id, "method": "tools/call", "params": map[string]any{"name": tool, "arguments": args}})
}
func serve(t *testing.T, tools []stdio.Tool) *runtime {
t.Setenv(stdio.ServedAs, "keeper")
inR, inW := io.Pipe()
outR, outW := io.Pipe()
go func() { _ = stdio.ServeOn("", tools, inR, outW) }()
t.Cleanup(func() { _ = inW.Close() })
return &runtime{t: t, lines: bufio.NewScanner(outR), in: inW}
}
// The bus of a moved provider maps its own bucket to the state `bound` and the assignments to the state
// `mesh-controller.assignments`, writes only the own bucket's state, and refuses any other bucket.
func TestStateBusMapsTheProvidersBucketsToTheRuntimesStates(t *testing.T) {
bus := StateBus("postgres")
rt := serve(t, []stdio.Tool{
{Name: "watch", Run: func(a map[string]any) (any, error) {
return nil, bus.Watch(context.Background(), a["bucket"].(string), func(string, []byte) {})
}},
{Name: "put", Run: func(map[string]any) (any, error) {
return "put", bus.Put(context.Background(), BucketOf("postgres"), "k", []byte(`{"a":1}`))
}},
{Name: "delete", Run: func(map[string]any) (any, error) {
return "deleted", bus.Delete(context.Background(), BucketOf("postgres"), "k")
}},
})
for i, c := range []struct{ bucket, state string }{
{BucketOf("postgres"), "bound"},
{AssignmentsBucket, "mesh-controller.assignments"},
} {
rt.call(i+1, "watch", map[string]any{"bucket": c.bucket})
w := rt.next()
if w["method"] != "mesh/state.watch" || w["params"].(map[string]any)["state"] != c.state {
t.Fatalf("bucket %s asked %v, want the state %s", c.bucket, w, c.state)
}
rt.send(map[string]any{"jsonrpc": "2.0", "id": w["id"], "result": map[string]any{}})
rt.next()
}
rt.call(3, "put", nil)
p := rt.next()
if p["method"] != "mesh/state.put" || p["params"].(map[string]any)["state"] != "bound" || p["params"].(map[string]any)["key"] != "k" {
t.Fatalf("put asked %v", p)
}
rt.send(map[string]any{"jsonrpc": "2.0", "id": p["id"], "result": map[string]any{"revision": 1}})
rt.next()
rt.call(4, "delete", nil)
d := rt.next()
if d["method"] != "mesh/state.delete" || d["params"].(map[string]any)["state"] != "bound" {
t.Fatalf("delete asked %v", d)
}
rt.send(map[string]any{"jsonrpc": "2.0", "id": d["id"], "result": map[string]any{}})
rt.next()
for _, bucket := range []string{"other_bound", "elsewhere"} {
if err := bus.Watch(context.Background(), bucket, func(string, []byte) {}); err == nil || !strings.Contains(err.Error(), "not a bucket") {
t.Fatalf("a watch of %s gave %v", bucket, err)
}
if err := bus.Put(context.Background(), bucket, "k", nil); err == nil {
t.Fatalf("a put to %s was allowed", bucket)
}
if err := bus.Delete(context.Background(), bucket, "k"); err == nil {
t.Fatalf("a delete in %s was allowed", bucket)
}
}
}
// A watch whose ctx has ended calls its handler no more, though the runtime keeps sending changes: the Bus contract
// says a watch lasts until ctx ends, and a stdio state watch lasts as long as the process.
func TestStateBusWatchCallsItsHandlerNoMoreOnceItsContextEnded(t *testing.T) {
bus := StateBus("postgres")
ctx, cancel := context.WithCancel(context.Background())
handled := make(chan string, 8)
rt := serve(t, []stdio.Tool{
{Name: "start", Run: func(map[string]any) (any, error) {
return "started", bus.Watch(ctx, AssignmentsBucket, func(key string, value []byte) {
handled <- key + "=" + string(value)
})
}},
{Name: "stop", Run: func(map[string]any) (any, error) { cancel(); return "stopped", nil }},
})
change := func(id int, watch any, key, op string) {
rt.send(map[string]any{"jsonrpc": "2.0", "id": id, "method": "mesh/state", "params": map[string]any{"watch": watch,
"change": map[string]any{"state": "mesh-controller.assignments", "key": key, "op": op, "value": "v"}}})
if r := rt.next(); r["id"] != float64(id) || r["error"] != nil {
t.Fatalf("a change was answered %v", r)
}
}
rt.call(1, "start", nil)
w := rt.next()
rt.send(map[string]any{"jsonrpc": "2.0", "id": w["id"], "result": map[string]any{}})
rt.next()
change(50, w["id"], "before", "put")
if got := <-handled; got != `before="v"` {
t.Fatalf("handled %q", got)
}
rt.call(2, "stop", nil)
rt.next()
change(51, w["id"], "after", "put")
change(52, w["id"], "after", "delete")
select {
case got := <-handled:
t.Fatalf("the handler was called after its context ended: %q", got)
case <-time.After(100 * time.Millisecond):
}
}