Files
mesh-sdk/go/provider/provider.go
T

838 lines
34 KiB
Go

// 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, issue 678): the provider module, not the controller, makes each bound
// consumer's entry.
//
// **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/base64"
"encoding/json"
"errors"
"fmt"
"regexp"
"sort"
"strings"
"sync"
"time"
stdio "git.novox.be/novox/mesh-sdk/go"
"git.novox.be/novox/mesh-sdk/go/provisions"
)
// 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
}
// ConditionalBus is a Bus that can also delete a key only while it still holds what the caller read. The helper uses
// it when the Bus offers it, so a pass that lets a consumer go never removes an entry another instance of the
// provider wrote after the pass read it; a Bus without it deletes unconditionally. DeleteIf returns an error that
// is stdio.ErrChanged when the key holds something else now.
type ConditionalBus interface {
Bus
DeleteIf(ctx context.Context, bucket, key string, held []byte) 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. A consumer's plan entry that names a
// credential form of its own wins over it (the plan's `credential-form`).
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"`
// VaultName is the name under which the vault keeps this consumer's credential, which the consumer's client asks
// the vault for: a name, never the credential. Its key is not "credential" because the runtime's state guard
// refuses any key ending in that word, whatever the value (issue 945).
VaultName string `json:"vault-name,omitempty"`
// Refused, when set, is why the consumer is not served, naming both ends.
Refused string `json:"refused,omitempty"`
// State is the provider's outcome for the consumer, which the holder of mesh-provisioner reads to set the
// provision's key (novox/hq ADR 0570 clause 3): provided, failed with Why, or waits with what is awaited. Empty from
// a provider before it.
State string `json:"state,omitempty"`
Why string `json:"why,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:"-"`
// CredentialForm and Name are from the consumer's plan entry (issue 825): the form of the credential the
// controller chose for this bind (`shared`, `pair`, `issued` or `given`) and the holder's name. When set, they
// win over the offer's form. Never written to the bucket.
CredentialForm, Name 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
entries map[string][]byte
// plans and provisions are the seat mesh-provisioner's states (ADR 0570 clauses 1, 2), by key.
plans map[string][]byte
provisions 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, entries: map[string][]byte{},
plans: map[string][]byte{}, provisions: map[string][]byte{}, said: map[string]string{}, poke: make(chan struct{}, 1)}
}
// ErrNotReady is what a Make returns, wrapped or alone, when the provider cannot make a consumer's provision yet but
// will without anyone's hand (its keys not read since a restart): the entry waits and keeps serving what it served.
// Any other error is a failure, and its entry serves nothing.
var ErrNotReady = errors.New("not ready yet")
// What a pass did for one bind.
const (
Answered = "answered"
Refused = "refused"
Waiting = "waiting"
Failed = "failed"
Removed = "removed"
// Superseded is an entry no bind asks for any more that another instance of the provider wrote after this pass
// read it: it is left standing.
Superseded = "superseded"
)
// 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)
}
for _, bucket := range []string{provisions.PlansBucket, provisions.ProvisionsBucket} {
into := p.plans
if bucket == provisions.ProvisionsBucket {
into = p.provisions
}
if err := p.watch(ctx, bucket, into); err != nil {
return fmt.Errorf("watching %s: %w", bucket, err)
}
}
tick := time.NewTicker(p.opts.Every)
defer tick.Stop()
for {
p.mu.Lock()
in := Inputs{Entries: clone(p.entries), Plans: clone(p.plans), Provisions: clone(p.provisions)}
p.mu.Unlock()
if _, err := p.passIn(ctx, in); 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 != BucketOf(p.opts.Module) {
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, or that
// is not the one its key names, is skipped and said (an outcome of status Failed under the assignment's key, and a
// line of the log), and the entries of its consumer are left as they are, since a bind missed would read as an
// unbind; the pass goes on for the rest. Run passes by itself; Pass is for a provider that passes when it chooses.
//
// Over StateBus each call to Pass leaves two watches open for the life of the process, silenced once the call has
// ended (see StateBus): a provider that passes often should use Run, which watches once.
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)
}
plans, err := read(provisions.PlansBucket)
if err != nil {
return nil, fmt.Errorf("reading %s: %w", provisions.PlansBucket, err)
}
provs, err := read(provisions.ProvisionsBucket)
if err != nil {
return nil, fmt.Errorf("reading %s: %w", provisions.ProvisionsBucket, err)
}
return p.passIn(ctx, Inputs{Entries: entries, Plans: plans, Provisions: provs})
}
// Inputs are what one pass reads: the provider's own entries, and the seat mesh-provisioner's plans and provisions
// (novox/hq ADR 0570 clauses 1, 2), each by key. The controller's assignments are read no more (clause 2).
type Inputs struct {
Entries, Plans, Provisions map[string][]byte
}
func (p *Provider) passIn(ctx context.Context, in Inputs) ([]Outcome, error) {
entries := in.Entries
var binds []Entry
var out []Outcome
unknown := map[string]bool{} // consumers whose assignment was skipped: their entries are not judged
// Every consumer is bound by its plan and its provision keys (ADR 0570 clauses 2, 3), and an entry no plan names is
// never let go by the provider itself (clause 5): only a retiring key with the operator's yes retires it.
seat := map[string]bool{}
gated := map[string]provisions.Provision{}
for _, key := range sortedKeys(in.Plans) {
plan, err := provisions.DecodePlan(in.Plans[key])
if err != nil || plan.Node+"."+plan.Module != key {
o := Outcome{Key: key, Status: Failed, Why: fmt.Sprintf("the plan %s is not readable, so who it binds is not "+
"known; it is skipped, and its entries are left as they are", key)}
out = append(out, o)
p.say(key, o.Status+" "+o.Why)
unknown[key], seat[key] = true, true
continue
}
seat[key] = true
for _, b := range plan.Plan.Provisions {
if b.Provider != p.opts.Node || b.ProviderModule != p.opts.Module {
continue
}
e := Entry{Provider: p.opts.Node, ProviderModule: p.opts.Module, Provision: b.Provision, Consumer: plan.Node,
Module: plan.Module, Local: b.Local, Contributes: b.Contributes, At: b.At, From: b.From, As: b.As,
Names: b.Names, CredentialForm: b.CredentialForm, Name: b.Name}
binds = append(binds, e)
}
}
for _, key := range sortedKeys(in.Provisions) {
if pr, err := provisions.Decode(in.Provisions[key]); err == nil && pr.Provider.Node == p.opts.Node &&
pr.Provider.Module == p.opts.Module {
gated[key] = pr
}
}
want := map[string]bool{}
for _, b := range binds {
key := b.Key()
want[key] = true
if why, ok := p.ready(key, gated, in.Provisions); !ok {
o := Outcome{Key: key, Status: Waiting, Why: why}
out = append(out, o)
p.say(key, o.Status+" "+o.Why)
continue
}
o := p.answer(ctx, b, entries[key])
out = append(out, o)
p.say(key, o.Status+" "+o.Why)
}
// ADR 0570 clause 5: a retiring key the operator said yes to is let go; one asked and not answered, or declined,
// keeps its entry.
for _, key := range sortedKeys(gated) {
if pr := gated[key]; pr.State == provisions.Retiring && pr.Warrant != nil && pr.Warrant.Option == "yes" {
want[key] = false
if _, held := entries[key]; held {
out = append(out, p.letGo(ctx, key, entries[key])...)
}
}
}
return out, nil
}
// consumerOf is the `<node>.<module>` an entry's key starts with: its first two parts.
func consumerOf(key string) string {
parts := strings.SplitN(key, ".", 3)
if len(parts) < 3 {
return key
}
return parts[0] + "." + parts[1]
}
// remove deletes an entry this instance read as held, and only while it is still as read when the Bus can say so.
func (p *Provider) remove(ctx context.Context, key string, held []byte) error {
if c, ok := p.opts.Bus.(ConditionalBus); ok {
return c.DeleteIf(ctx, BucketOf(p.opts.Module), key, held)
}
return p.opts.Bus.Delete(ctx, BucketOf(p.opts.Module), key)
}
// 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 != "" {
e.State, e.Why = provisions.Failed, 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 = ""
e.State, e.Why = provisions.Failed, e.Refused
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
// The plan's form wins; the offer's is the default.
// An offer that declares no credential has none whatever the plan says: the controller writes "pair" for it.
form := e.CredentialForm
if offer.Credential == NoCredential {
form = ""
} else if form == "" {
form = map[CredentialForm]string{Shared: "shared", PerConsumer: "pair"}[offer.Credential]
}
switch form {
case "shared":
e.User = offer.User
e.VaultName = SharedName(e.Provider, e.Provision)
case "pair":
e.User = e.As
e.VaultName = PairName(e.Provider, e.Provision, e.Consumer, e.Module, e.Local)
case "issued", "given":
// A per-holder value, as mesh-vault names it (perholder.go).
e.User = e.As
e.VaultName = HolderName(e.Provider, e.Provision, e.Consumer, e.Module)
}
if e.VaultName != "" {
credential, err = p.opts.Vault.Value(ctx, e.VaultName)
if err != nil {
return Outcome{Key: key, Status: Waiting, Why: fmt.Sprintf("the vault's %s: %v", e.VaultName, err)}
}
}
if p.opts.Make != nil {
if err := p.opts.Make(ctx, Consumer{Entry: e, Credential: credential}); errors.Is(err, ErrNotReady) {
// **Not ready is no failure** (the regression of 2026-10-30: gitea after a restart, its keys not read yet,
// stopped serving every consumer): the entry says waits with why, and keeps what the last entry served, so a
// consumer goes on with it until the provider can make it again.
why := "the provider is not ready to make it yet: " + redact(err.Error(), credential)
waiting := e
var prev Entry
if len(have) > 0 && json.Unmarshal(have, &prev) == nil && prev.State == provisions.Provided {
waiting.Values, waiting.VaultName, waiting.User = prev.Values, prev.VaultName, prev.User
} else {
waiting.Values, waiting.VaultName, waiting.User = nil, "", ""
}
waiting.State, waiting.Why = provisions.Waits, why
p.write(ctx, waiting, have, Outcome{Key: key, Status: Waiting, Why: why})
return Outcome{Key: key, Status: Waiting, Why: why}
} else if err != nil {
why := "the provider could not make it: " + redact(err.Error(), credential)
// Said in its entry, which the holder of mesh-provisioner reads (ADR 0570 clause 3), with nothing served.
failed := e
failed.Values, failed.VaultName, failed.User, failed.State, failed.Why = nil, "", "", provisions.Failed, why
p.write(ctx, failed, have, Outcome{Key: key, Status: Failed, Why: why})
return Outcome{Key: key, Status: Failed, Why: why}
}
}
e.State = provisions.Provided
return p.write(ctx, e, have, Outcome{Key: key, Status: Answered})
}
// redact takes a credential out of text in the forms it can come back in: as it is, in base64 (standard and URL,
// padded and not) and as a JSON string says it (escapes included).
func redact(text string, credential []byte) string {
if len(credential) == 0 {
return text
}
quoted, _ := json.Marshal(string(credential))
forms := []string{string(credential), base64.StdEncoding.EncodeToString(credential),
base64.RawStdEncoding.EncodeToString(credential), base64.URLEncoding.EncodeToString(credential),
base64.RawURLEncoding.EncodeToString(credential), string(quoted[1 : len(quoted)-1])}
sort.SliceStable(forms, func(i, j int) bool { return len(forms[i]) > len(forms[j]) })
for _, f := range forms {
text = strings.ReplaceAll(text, f, "<credential>")
}
return text
}
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 provisions.VaultName(provisions.FormShared, provisions.Provision{Provision: provision, Provider: provisions.End{Node: provider}})
}
// HolderName is the vault's value of one holder's own credential, issued or given for it (mesh-vault perholder.go).
func HolderName(provider, provision, node, module string) string {
return provisions.VaultName(provisions.FormIssued, provisions.Provision{Provision: provision,
Provider: provisions.End{Node: provider}, Consumer: provisions.End{Node: node, Module: module}})
}
// 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 {
return provisions.VaultName(provisions.FormPair, provisions.Provision{Provision: provision, Local: local,
Provider: provisions.End{Node: provider}, Consumer: provisions.End{Node: node, Module: module}})
}
// 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 provisions.PlansBucket and provisions.ProvisionsBucket are the
// states it reads, `mesh-provisioner.plans` and `mesh-provisioner.provisions` (each 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 provisions.PlansBucket:
return stdio.State(provisions.PlansRead), nil
case provisions.ProvisionsBucket:
return stdio.State(provisions.ProvisionsRead), 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
}
// DeleteIf removes the key only while it still holds what the caller read; it reads the key first, so another
// writer's change in that short while can still be lost.
func (b stateBus) DeleteIf(_ context.Context, bucket, key string, held []byte) error {
s, err := b.state(bucket)
if err != nil {
return err
}
cur, err := s.Get(key)
if err != nil {
return err
}
if cur == nil {
return nil
}
if !sameJSON(cur.Value, held) {
return fmt.Errorf("%s is not as it was read: %w", key, stdio.ErrChanged)
}
return s.Delete(key)
}
func (b stateBus) Delete(_ context.Context, bucket, key string) error {
s, err := b.state(bucket)
if err != nil {
return err
}
return s.Delete(key)
}
// Standing is whether an assignment's value is a standing assignment: false when `ended` is set (the module was
// taken off its node, novox/hq ADR 0408 clause 4) or `state` is not empty (a pending one). Such a value is read
// exactly as a removed key. Every other field, known or not, changes nothing (issue 1098).
func Standing(value []byte) bool {
var v struct {
Ended json.RawMessage `json:"ended"`
State json.RawMessage `json:"state"`
}
if json.Unmarshal(value, &v) != nil {
return true // not decided here: the parse that follows refuses it
}
return !fieldSet(v.Ended) && !fieldSet(v.State)
}
// fieldSet is whether a field carries something: not absent, null or empty.
func fieldSet(f json.RawMessage) bool {
s := strings.TrimSpace(string(f))
return s != "" && s != "null" && s != `""`
}
// letGo lets a consumer's entry go: the provider hears it gone (Gone, which retires what it kept for the consumer), and
// the entry is removed while it is still as read.
func (p *Provider) letGo(ctx context.Context, key string, held []byte) []Outcome {
var e Entry
if err := json.Unmarshal(held, &e); err != nil || e.Provider != p.opts.Node || e.ProviderModule != p.opts.Module {
return nil // another machine's instance of this provider wrote it, or nothing of this helper's
}
said := func(o Outcome) []Outcome {
p.say(key, o.Status+" "+o.Why)
return []Outcome{o}
}
if p.opts.Gone != nil {
if err := p.opts.Gone(ctx, e); err != nil {
return said(Outcome{Key: key, Status: Failed, Why: "no longer bound, and the provider could not let it go: " + err.Error()})
}
}
if err := p.remove(ctx, key, held); err != nil {
if errors.Is(err, stdio.ErrChanged) {
return said(Outcome{Key: key, Status: Superseded, Why: "another instance of the provider wrote its entry " +
"since this pass read it, so it stands"})
}
return said(Outcome{Key: key, Status: Failed, Why: "no longer bound, and its entry could not be removed: " + err.Error()})
}
p.say(key, Removed)
p.mu.Lock()
delete(p.said, key)
p.mu.Unlock()
return []Outcome{{Key: key, Status: Removed}}
}
// ready says whether the provision under key may be made now (novox/hq ADR 0570 clause 3): its key is on the seat, it is
// asked or provided, and every provision it depends on is provided. Otherwise why it waits, naming the key or the
// dependency.
func (p *Provider) ready(key string, mine map[string]provisions.Provision, all map[string][]byte) (string, bool) {
pr, ok := mine[key]
if !ok {
return "mesh-provisioner holds no key " + key + " naming this provider yet", false
}
if pr.State != provisions.Asked && pr.State != provisions.Provided {
return "its key is " + pr.State + ": " + pr.Why, false
}
for _, d := range pr.DependsOn {
dep, err := provisions.Decode(all[d])
if err != nil {
return "it depends on " + d + ", which mesh-provisioner holds no readable key of", false
}
if dep.State != provisions.Provided {
return "it depends on " + d + ", which is " + dep.State, false
}
}
return "", true
}