novox/mesh-sdk
13
0
Fork 0
Code Issues 1 Pull Requests Actions Packages Projects Releases Wiki Activity

Merge pull request 'Feature providers-after-switch-on, step 1: the provider helper's fixes and the moved-provider harness (issues 945, 825, 849)' (#31) from feat/providers-after-switch-on into main

Browse Source
This commit was merged in pull request #31.
This commit is contained in:
mesh-tools
2026-10-16 19:13:11 +00:00
parent 219bedbdd2 a4e7f480b3
commit f70a99dbf4
13 changed files with 3492 additions and 46 deletions
Download Patch File Download Diff File Expand all files Collapse all files
go/provider/harness/ctx_test.go
+5
View File
@@ -0,0 +1,5 @@
package harness
import "context"
var ctx = context.Background()
go/provider/harness/early_test.go
+62
View File
@@ -0,0 +1,62 @@
package harness
// The moved path started before its runtime answers asks again until it does (issue 943).
import (
"context"
"errors"
"testing"
"time"
"git.novox.be/novox/mesh-sdk/go/provider"
)
// earlyBus refuses its first watches as the runtime does before Serve runs, then answers as the bucket bus does.
type earlyBus struct {
*bucketBus
refuse int
}
func (b *earlyBus) Watch(ctx context.Context, bucket string, handle func(string, []byte)) error {
if b.refuse > 0 {
b.refuse--
return errors.New("before Serve: a bundle reaches the bus through the runtime that launched it")
}
return b.bucketBus.Watch(ctx, bucket, handle)
}
func TestAMovedPathStartedTooEarlyAsksAgainUntilTheRuntimeAnswers(t *testing.T) {
w := newMovedWorld(t, map[[2]string]string{})
w.run.FirstRetry = time.Millisecond
w.run.loaded = false
bus := &earlyBus{bucketBus: w.bus, refuse: 3}
p := provider.New(provider.Options{Node: "novox", Module: "postgres", Bus: w.bus,
Vault: provider.ConsumersFile{Paths: []string{w.file}}, Make: w.run.Make, Gone: w.run.Gone,
Every: time.Hour, Logf: func(string, ...any) {}})
ctx, cancel := context.WithCancel(context.Background())
done := make(chan error, 1)
go func() { done <- w.run.Run(ctx, p, bus) }()
deadline := time.After(time.Second)
for {
w.run.mu.Lock()
loaded := w.run.loaded
w.run.mu.Unlock()
if loaded {
break
}
select {
case err := <-done:
cancel()
t.Fatalf("Run stopped before the runtime answered: %v", err)
case <-deadline:
cancel()
t.Fatal("the plans were never watched")
case <-time.After(time.Millisecond):
}
}
cancel()
<-done
if bus.refuse != 0 {
t.Fatalf("%d refusals left", bus.refuse)
}
}
go/provider/harness/grants_moved_test.go
+123
View File
@@ -0,0 +1,123 @@
package harness
// postgres handed to the provisioner still serves the consumers the controller's grant file gives (issue 937): a
// consumer not moved has no plan naming postgres, only its grant, until R1 (issue 682).
import (
"encoding/json"
"os"
"path/filepath"
"reflect"
"testing"
"time"
)
// granting gives the moved world a grant file, as the controller writes it for the consumers not moved.
func (w *movedWorld) granting(given ...map[string]any) {
if w.h.Receives == "" {
w.h.Receives = filepath.Join(w.t.TempDir(), "mesh.json")
}
dir := filepath.Dir(w.h.Receives)
for _, g := range given {
secret := filepath.Join(dir, g["as"].(string)+".secret")
if err := os.WriteFile(secret, []byte("pw-"+g["as"].(string)+"\n"), 0o600); err != nil {
w.t.Fatal(err)
}
g["secret"] = secret
}
if given == nil {
given = []map[string]any{}
}
raw, _ := json.Marshal(map[string]any{"requirement": ProvisionName, "given": given})
if err := os.WriteFile(w.h.Receives, raw, 0o600); err != nil {
w.t.Fatal(err)
}
}
// settled is enough passes and time for a retirement to be stable.
func (w *movedWorld) settled() {
for i := 0; i < 8; i++ {
w.pass()
}
w.now = w.now.Add(15 * time.Minute)
w.pass()
w.pass()
}
// The logins the grant file gives are asked for: made as the grant path makes them, never offered for retirement,
// beside a moved consumer the helper makes from its plan.
func TestAConsumerOnlyTheGrantFileGivesIsAskedForAndMade(t *testing.T) {
bound, logins := bindings(1)
w := newMovedWorld(t, bound)
w.a.inv.Active = []string{"mesh_ace_baserow", "mesh_ace_letta", logins[0]}
w.bus.bind("ace", "app1", nil)
w.granting(
map[string]any{"as": "mesh_ace_baserow", "node": "ace", "values": map[string]any{"name": "baserow"}},
map[string]any{"as": "mesh_ace_letta", "node": "ace",
"values": map[string]any{"name": "letta", "extensions": []any{"vector"}}},
)
w.settled()
if len(w.a.removed) != 0 || w.h.r.waiting != nil {
t.Fatalf("retired %v, waiting %+v: the grant file's logins were counted unasked", w.a.removed, w.h.r.waiting)
}
made := map[string]Provision{}
for _, c := range w.a.created {
made[c.As] = c
}
letta := made["mesh_ace_letta"]
if letta.Password != "pw-mesh_ace_letta" || letta.Consumer != "ace" ||
!reflect.DeepEqual(letta.Values, map[string]any{"name": "letta", "extensions": []any{"vector"}}) {
t.Fatalf("made %+v for letta, as its grant gives it", letta)
}
if _, ok := made["mesh_ace_baserow"]; !ok {
t.Fatalf("baserow not made: %+v", w.a.created)
}
if _, ok := made[logins[0]]; !ok {
t.Fatalf("the moved consumer not made: %+v", w.a.created)
}
}
// A login both a plan and the grant file name is the plan's: made once, with the plan's credential.
func TestALoginAPlanNamesIsNotMadeFromTheGrantFile(t *testing.T) {
bound, logins := bindings(1)
w := newMovedWorld(t, bound)
w.bus.bind("ace", "app1", nil)
w.granting(map[string]any{"as": logins[0], "node": "ace", "values": map[string]any{"name": "app1"}})
w.pass()
w.pass()
for _, c := range w.a.created {
if c.As == logins[0] && c.Password != "pw-app1" {
t.Fatalf("made %s with the grant's password %q, not the plan's", c.As, c.Password)
}
}
}
// An unreadable grant file is not nobody asking (issue 241): nothing is retired while it cannot be read.
func TestAnUnreadableGrantFileRetiresNothing(t *testing.T) {
bound, logins := bindings(1)
w := newMovedWorld(t, bound)
w.a.inv.Active = []string{"mesh_ace_baserow", logins[0]}
w.bus.bind("ace", "app1", nil)
w.h.Receives = filepath.Join(t.TempDir(), "absent.json")
w.settled()
if len(w.a.removed) != 0 || w.h.r.waiting != nil {
t.Fatalf("retired %v, waiting %+v with the grant file unreadable", w.a.removed, w.h.r.waiting)
}
}
// A login the grant file stops giving is retired by the usual rules: stable, then retired, never dropped.
func TestALoginTheGrantFileStopsGivingIsRetired(t *testing.T) {
bound, _ := bindings(1)
w := newMovedWorld(t, bound)
w.bus.bind("ace", "app1", nil)
w.granting(
map[string]any{"as": "mesh_ace_baserow", "node": "ace", "values": map[string]any{"name": "baserow"}},
map[string]any{"as": "mesh_ace_letta", "node": "ace", "values": map[string]any{"name": "letta"}},
)
w.pass()
w.granting(map[string]any{"as": "mesh_ace_baserow", "node": "ace", "values": map[string]any{"name": "baserow"}})
w.settled()
if !reflect.DeepEqual(w.a.removed, []string{"mesh_ace_letta"}) || len(w.a.deleted) != 0 {
t.Fatalf("retired %v, deleted %v", w.a.removed, w.a.deleted)
}
}
go/provider/harness/harness.go
+572
View File
@@ -0,0 +1,572 @@
// Package harness is the reconcile loop and the retirement of a provider that keeps a resource per consumer; the
// module supplies an Adapter.
package harness
// The reconcile loop every provider shares, as the TypeScript SDK's runProvisioner runs it
// (@novox/mesh-sdk/provisioner, 0.1.12), line for line in behaviour. It moved here from the postgres and keycloak
// providers, which each carried an identical copy (novox/hq ADR 0039: the loop is the SDK's, the adapter is the
// module's; issue 849).
//
// Read the contributions the mesh delivered; bring each consumer's resource into being through the
// adapter, under the login and password the mesh minted; retire what the mesh no longer asks for.
// **A provider creates the credential the mesh minted, and seals nothing (novox/hq ADR 0048).**
//
// **A provider that keeps failing a consumer says so on the bus (novox/hq ADR 0224).** A consumer
// whose create, check or secret has failed without one success in between for FailingAfter is
// announced as `provisioner.failing` — naming the consumer, its machine and the class of error — and
// again every SayAgainEvery while it lasts; the first success after that is `provisioner.recovered`.
// The controller keeps the newest per provider and consumer and `status` names it. On 2026-10-05 the
// identity provider failed every consumer 31,000 times in a day and said so only in its journal
// (novox/hq issue 179).
//
// **A consumer the mesh stops asking for is retired, not withdrawn, and deleted only by a person
// (novox/hq ADR 0230, the operator's model of 2026-10-06).** Retiring disables its access — reversibly —
// and marks its login and data "to delete" with when and why; nothing is deleted. Asked for again, the
// ordinary create re-enables it as it was. It is retired only once the same set has gone unasked in
// StablePasses consecutive passes and for StableFor, and a set larger than the bound waits for a
// person. retirement.go.
import (
"context"
"encoding/json"
"errors"
"fmt"
"net/url"
"os"
"strings"
"sync"
"time"
)
// Provision is one consumer's resource to bring into being — everything the mesh derived and delivered.
type Provision struct {
// As is the login the mesh derived and gave the consumer to present.
As string
// Password is the one the mesh minted, read from the file the host unsealed.
Password string
// Values are what the consumer contributed (e.g. {"name": "letta", "extensions": ["vector"]}).
Values map[string]any
// Derived is what this provider's own definition derives for the consumer (novox/hq ADR 0201).
Derived map[string]any
// At is where the consumer is; Consumer is its node.
At string
Consumer string
}
// Adapter is the per-service half.
type Adapter interface {
// Create brings the consumer into being under the mesh's login and password — and, for one that
// was retired, re-enables it as it was and clears its mark to delete.
Create(ctx context.Context, p Provision) error
// Retire disables the consumer's access, reversibly, and marks its login and data to delete with
// when and why. It deletes nothing (novox/hq ADR 0230). derived is what the mesh last derived,
// remembered here; nil for a consumer this process did not make.
Retire(ctx context.Context, as string, derived map[string]any, why string, at time.Time) error
// Holds says whether the backend still holds the consumer exactly as p says. Read-only.
Holds(ctx context.Context, p Provision) (bool, error)
// Inventory is what the backend holds that the mesh made: consumers active and retired, read from
// the backend itself so a restart forgets nothing. Read-only.
Inventory(ctx context.Context) (Inventory, error)
// Delete removes one retired consumer — its login and its data — for good. Only ever asked by a
// person, through `cleanup delete`; the harness has checked it is retired and not asked for.
Delete(ctx context.Context, r Retired) (freedBytes int64, err error)
}
// Harness is the loop's settings and memory.
type Harness struct {
Resource string
Receives string
Adapter Adapter
Every time.Duration // 5s
VerifyEvery time.Duration // 60s
HoldsTimeout time.Duration // 30s
Log func(format string, args ...any)
Now func() time.Time
// Announce publishes one of the provider's standing events; nil announces nothing. Node is the
// machine this provider runs on, said in each.
Announce func(event string, body map[string]any)
Node string
// FailingAfter is how long a consumer fails without a success before it is announced (5m);
// SayAgainEvery is how often it is announced again while it lasts (15m), so a controller that
// missed the first hears the next, and a standing nobody repeats can be told from one that holds.
FailingAfter time.Duration
SayAgainEvery time.Duration
// StablePasses (5) is how many consecutive passes must see the same set no longer asked for, and
// StableFor (10m) how long it must have held since the first of them, before it is retired. RetireAtOnce (3) and RetireFraction (0.5) bound what is retired without a person:
// more consumers than RetireAtOnce, or — where more than one is held — a larger share of those held
// than RetireFraction, waits for `retire approve` (novox/hq ADR 0230).
StablePasses int
StableFor time.Duration
RetireAtOnce int
RetireFraction float64
// mu is held by a pass and by every tool a person asks, so the two never interleave.
mu sync.Mutex
verifiedAt time.Time
r retirement
applied map[string]appliedEntry
lost map[string]brake
waiting map[string]int
failing map[string]failure
trouble map[string]*standing
cleared map[string]bool
lastWarning string
}
// standing is one consumer's unbroken run of failures: since when, how often, and the last error.
type standing struct {
node string
since time.Time
attempts int
class string
text string
saidAt time.Time
}
// The events a provider's standing is announced as (novox/hq ADR 0224). The controller derives the
// permission to emit them for every module that receives contributions; no manifest lists them.
const (
EventFailing = "provisioner.failing"
EventRecovered = "provisioner.recovered"
)
// The classes of error a standing is announced with: what a person reading `status` needs to know
// before reading the journal. An adapter may say better (Classifier).
const (
ClassCredentials = "credentials-rejected"
ClassUnreachable = "unreachable"
ClassSecret = "secret-unreadable"
ClassRefused = "refused"
)
// Classifier is an adapter that can say what class an error of its own is.
type Classifier interface {
Class(err error) string
}
type appliedEntry struct {
hash string
derived map[string]any
node string
}
type brake struct {
times int
nextAt time.Time
}
type failure struct {
text string
times int
}
// The longest a consumer whose create keeps failing to satisfy holds waits between checks.
const maxBackoff = time.Hour
// How many passes a secret may be unreadable before it stops being called a race (issue 225), and
// once said loudly, how often it is repeated. The same cadence quiets a create that keeps failing
// the same way.
const (
patiently = 12
loudlyEvery = 240
)
type contribution struct {
As string `json:"as"`
Secret string `json:"secret"`
Node string `json:"node"`
At string `json:"at"`
Values map[string]any `json:"values"`
Derived map[string]any `json:"derived"`
}
func (h *Harness) init() {
if h.Every == 0 {
h.Every = 5 * time.Second
}
if h.VerifyEvery == 0 {
h.VerifyEvery = time.Minute
}
if h.HoldsTimeout == 0 {
h.HoldsTimeout = 30 * time.Second
}
if h.Now == nil {
h.Now = time.Now
}
if h.FailingAfter == 0 {
h.FailingAfter = 5 * time.Minute
}
if h.SayAgainEvery == 0 {
h.SayAgainEvery = 15 * time.Minute
}
if h.StablePasses == 0 {
h.StablePasses = 5
}
if h.StableFor == 0 {
h.StableFor = StableFor
}
if h.RetireAtOnce == 0 {
h.RetireAtOnce = 3
}
if h.RetireFraction == 0 {
h.RetireFraction = 0.5
}
if h.Log == nil {
h.Log = func(format string, args ...any) { fmt.Fprintf(os.Stderr, format+"\n", args...) }
}
if h.applied == nil {
h.applied = map[string]appliedEntry{}
h.lost = map[string]brake{}
h.waiting = map[string]int{}
h.failing = map[string]failure{}
h.trouble = map[string]*standing{}
h.cleared = map[string]bool{}
h.r.retired = map[string]retiredEntry{}
}
}
// Run reconciles until ctx ends. One consumer's failure never stops the others'.
func (h *Harness) Run(ctx context.Context) {
h.init()
for {
h.Reconcile(ctx)
select {
case <-ctx.Done():
return
case <-time.After(h.Every):
}
}
}
func (h *Harness) say(format string, args ...any) {
h.Log("[provisioner:"+h.Resource+"] "+format, args...)
}
func (h *Harness) warn(why string) {
if why == h.lastWarning {
return
}
if why != "" {
h.say("%s; nothing applied or removed until it can be read", why)
} else {
h.say("contributions file readable again")
}
h.lastWarning = why
}
// readContributions answers the consumers asked for, or nil when the file says nothing usable.
// **Nothing read is not nobody asking** (novox/hq issue 241): only a file that was read can retire anything.
func (h *Harness) readContributions() []contribution {
raw, err := os.ReadFile(h.Receives)
if err != nil {
h.warn(fmt.Sprintf("contributions file unreadable (%s): %v", h.Receives, err))
return nil
}
var doc struct {
Requirement string `json:"requirement"`
Given json.RawMessage `json:"given"`
}
if err := json.Unmarshal(raw, &doc); err != nil {
h.warn(fmt.Sprintf("contributions file is not JSON (%s): %v", h.Receives, err))
return nil
}
if doc.Requirement != "" && doc.Requirement != h.Resource {
h.warn(fmt.Sprintf("%s is for %s, not %s", h.Receives, doc.Requirement, h.Resource))
return nil
}
var given []contribution
if len(doc.Given) == 0 || string(doc.Given) == "null" || json.Unmarshal(doc.Given, &given) != nil {
h.warn(fmt.Sprintf("%s has no given list", h.Receives))
return nil
}
h.warn("")
out := []contribution{}
for _, g := range given {
// No `as` is not a credential grant: nothing to create for it.
if g.As != "" && g.Secret != "" {
out = append(out, g)
}
}
return out
}
func hashOf(as, password string, values, derived map[string]any) string {
// Derived is in the hash: a provider that renames what it derives gave a different resource.
b, _ := json.Marshal([]any{as, password, orEmpty(values), orEmpty(derived)})
return string(b)
}
func orEmpty(m map[string]any) map[string]any {
if m == nil {
return map[string]any{}
}
return m
}
// Reconcile is one pass.
func (h *Harness) Reconcile(ctx context.Context) {
h.mu.Lock()
defer h.mu.Unlock()
h.init()
given := h.readContributions()
if given == nil {
// Not a result: the passes that must agree before anything is retired start again.
h.r.key, h.r.count = "", 0
return
}
want := map[string]bool{}
for _, g := range given {
want[g.As] = true
}
h.seed(ctx, want)
h.apply(ctx, given)
// Retire what the mesh has stably stopped asking for — within the bound, or with a person.
h.retireUnasked(ctx, want)
h.r.lastWant = want
for as := range h.failing {
if !want[as] {
delete(h.failing, as)
}
}
// A consumer the mesh stopped asking for that this provider holds nothing for is no longer failed by
// anyone: said, so a standing the controller keeps for it is cleared rather than left naming a
// consumer that is gone. One still held is said recovered when it is retired.
for as := range h.trouble {
if _, held := h.applied[as]; !want[as] && !held {
h.recovered(as, "no longer asked for")
}
}
}
// apply makes each consumer the grant file gives, and checks every VerifyEvery that the backend still holds it. The
// caller holds h.mu. One consumer's failure never stops the others'.
func (h *Harness) apply(ctx context.Context, given []contribution) {
verifying := h.Now().Sub(h.verifiedAt) >= h.VerifyEvery
if verifying {
h.verifiedAt = h.Now()
}
for _, g := range given {
raw, err := os.ReadFile(g.Secret)
if err != nil {
// A secret the host has not written yet is a race on the first pass; past a minute it is
// a person's to look at, and said so (novox/hq issue 225).
n := h.waiting[g.As] + 1
h.waiting[g.As] = n
if n <= patiently {
h.say("%s: secret not readable yet (%s): %v", g.As, g.Secret, err)
} else if n == patiently+1 || n%loudlyEvery == 0 {
h.say("%s: CANNOT READ the secret after %d attempts (%s): %v. This is not a race any more — "+
"nothing has been provisioned for this consumer and nothing will be until somebody looks. "+
"Check who owns the file and who this process runs as (novox/hq issue 225)", g.As, n, g.Secret, err)
}
h.failed(g.As, g.Node, ClassSecret, fmt.Sprintf("secret not readable (%s): %v", g.Secret, err))
continue
}
delete(h.waiting, g.As)
password := strings.TrimSuffix(string(raw), "\n")
p := Provision{As: g.As, Password: password, Values: orEmpty(g.Values), Derived: orEmpty(g.Derived), At: g.At, Consumer: g.Node}
hash := hashOf(g.As, password, g.Values, g.Derived)
reapplying := 0
if was, ok := h.applied[g.As]; ok && was.hash == hash {
if !verifying {
continue
}
b, braked := h.lost[g.As]
if braked && h.Now().Before(b.nextAt) {
continue
}
hctx, cancel := context.WithTimeout(ctx, h.HoldsTimeout)
held, err := h.Adapter.Holds(hctx, p)
timedOut := errors.Is(hctx.Err(), context.DeadlineExceeded)
cancel()
if err != nil {
// Unable to ask is not evidence of loss. A backend that timed out will time out for
// the next consumer too, so the rest of this pass is not asked.
text := scrub(err, password)
h.say("%s: could not check the backend, will ask again: %s", g.As, text)
h.failed(g.As, g.Node, h.classOf(err, text), text)
if timedOut {
verifying = false
}
continue
}
if held {
delete(h.lost, g.As)
h.succeeded(g.As)
continue
}
reapplying = b.times + 1
if reapplying == 1 {
h.say("%s: the backend no longer holds it; applying again", g.As)
} else {
h.say("%s: still not held after being applied again (%d times in a row) — create does not "+
"produce what holds checks; applying again", g.As, reapplying)
}
}
if err := h.Adapter.Create(ctx, p); err != nil {
text := scrub(err, password)
f := h.failing[g.As]
if f.text != text {
f = failure{text: text}
}
f.times++
h.failing[g.As] = f
// Said each time it changes, and while it stays the same, as rarely as a lost secret.
if f.times == 1 || f.times%loudlyEvery == 0 {
h.say("%s: create failed, will retry: %s", g.As, text)
}
h.failed(g.As, g.Node, h.classOf(err, text), text)
if reapplying > 0 {
h.lost[g.As] = brake{times: reapplying - 1}
}
continue
}
if f, was := h.failing[g.As]; was {
h.say("%s: created, after %d failed attempt(s)", g.As, f.times)
delete(h.failing, g.As)
}
h.succeeded(g.As)
h.applied[g.As] = appliedEntry{hash: hash, derived: p.Derived, node: g.Node}
h.reenabled(g.As, g.Node)
if reapplying == 0 {
delete(h.lost, g.As)
} else {
wait := h.VerifyEvery << (reapplying - 1)
if wait > maxBackoff || wait <= 0 {
wait = maxBackoff
}
h.lost[g.As] = brake{times: reapplying, nextAt: h.Now().Add(wait)}
if reapplying > 1 {
h.say("%s: next check in %s", g.As, wait.Round(time.Second))
}
}
}
}
// failed counts one more failure in a consumer's unbroken run, and announces the run once it has
// lasted FailingAfter — then again every SayAgainEvery while it lasts.
func (h *Harness) failed(as, node, class, text string) {
now := h.Now()
s := h.trouble[as]
if s == nil {
s = &standing{since: now}
h.trouble[as] = s
}
s.node, s.class, s.text = node, class, text
s.attempts++
if now.Sub(s.since) < h.FailingAfter {
return
}
if !s.saidAt.IsZero() && now.Sub(s.saidAt) < h.SayAgainEvery {
return
}
first := s.saidAt.IsZero()
s.saidAt = now
if first {
h.say("%s: FAILING for %s (%d attempts, %s): %s. Announced as %s; `status` names it until it "+
"succeeds (novox/hq ADR 0224)", as, now.Sub(s.since).Round(time.Second), s.attempts, class, text, EventFailing)
}
h.announce(EventFailing, map[string]any{
"provider": h.Resource, "provider-node": h.Node,
"consumer": as, "node": node,
"class": class, "error": clip(text),
"since": s.since.UTC().Format(time.RFC3339), "attempts": s.attempts,
})
}
// succeeded ends a consumer's run of failures; one that was announced is announced recovered.
//
// **And the first success for a consumer since this process started is announced too**, failing or
// not: a provider that announced a failure and was restarted has forgotten it, and without this the
// controller would name the consumer failing for ever after it recovered unheard.
func (h *Harness) succeeded(as string) {
if h.trouble[as] == nil && !h.cleared[as] {
h.cleared[as] = true
h.announce(EventRecovered, map[string]any{
"provider": h.Resource, "provider-node": h.Node, "consumer": as, "why": "first-success",
})
return
}
h.cleared[as] = true
h.recovered(as, "")
}
func (h *Harness) recovered(as, why string) {
s := h.trouble[as]
if s == nil {
return
}
delete(h.trouble, as)
if s.saidAt.IsZero() {
return // never announced, so there is nothing to take back
}
if why == "" {
h.say("%s: recovered after %s and %d failed attempt(s)", as, h.Now().Sub(s.since).Round(time.Second), s.attempts)
}
body := map[string]any{
"provider": h.Resource, "provider-node": h.Node, "consumer": as, "node": s.node,
"since": s.since.UTC().Format(time.RFC3339), "attempts": s.attempts,
}
if why != "" {
body["why"] = why
}
h.announce(EventRecovered, body)
}
func (h *Harness) announce(event string, body map[string]any) {
if h.Announce != nil {
h.Announce(event, body)
}
}
// classOf is an error's class: the adapter's word when it has one, else read from the text.
func (h *Harness) classOf(err error, text string) string {
if c, ok := h.Adapter.(Classifier); ok {
if class := c.Class(err); class != "" {
return class
}
}
return ClassOf(text)
}
// ClassOf reads an error's class from its text — the words the backends the mesh runs use.
func ClassOf(text string) string {
t := strings.ToLower(text)
for _, w := range []string{"invalid_grant", "invalid user credentials", "password authentication failed",
"authentication failed", "unauthorized", " 401"} {
if strings.Contains(t, w) {
return ClassCredentials
}
}
for _, w := range []string{"connection refused", "no such host", "i/o timeout", "deadline exceeded",
"connection reset", "network is unreachable", "no route to host", "eof"} {
if strings.Contains(t, w) {
return ClassUnreachable
}
}
return ClassRefused
}
// clip keeps an announced error to what belongs in a status line.
func clip(text string) string {
const most = 300
if len(text) <= most {
return text
}
return text[:most] + "…"
}
// scrub is an error's text with the consumer's password removed, raw and URL-encoded.
func scrub(err error, password string) string {
text := err.Error()
if password == "" {
return text
}
for _, form := range []string{password, url.QueryEscape(password), url.PathEscape(password)} {
text = strings.ReplaceAll(text, form, "***")
}
return text
}
go/provider/harness/harness_test.go
+154
View File
@@ -0,0 +1,154 @@
package harness
import (
"encoding/json"
"os"
"path/filepath"
"strings"
"testing"
"time"
)
type world struct {
t *testing.T
dir string
receives string
now time.Time
h *Harness
a *recorder
said []string
}
func newWorld(t *testing.T) *world {
w := &world{t: t, dir: t.TempDir(), now: time.Date(2026, 10, 5, 12, 0, 0, 0, time.UTC), a: &recorder{held: true}}
w.receives = filepath.Join(w.dir, "mesh.json")
w.h = &Harness{Resource: "postgres-database", Receives: w.receives, Adapter: w.a,
Now: func() time.Time { return w.now },
Log: func(f string, a ...any) { w.said = append(w.said, f) }}
return w
}
func (w *world) give(given ...map[string]any) {
for _, g := range given {
secret := filepath.Join(w.dir, g["as"].(string)+".secret")
if err := os.WriteFile(secret, []byte("pw-"+g["as"].(string)+"\n"), 0o600); err != nil {
w.t.Fatal(err)
}
g["secret"] = secret
}
if given == nil {
given = []map[string]any{}
}
raw, _ := json.Marshal(map[string]any{"requirement": "postgres-database", "given": given})
if err := os.WriteFile(w.receives, raw, 0o600); err != nil {
w.t.Fatal(err)
}
}
func TestAConsumerIsCreatedOnceUnderTheMeshsLoginAndPassword(t *testing.T) {
w := newWorld(t)
w.give(map[string]any{"as": "mesh_ace_letta", "node": "ace", "values": map[string]any{"name": "letta"}})
w.h.Reconcile(ctx)
w.h.Reconcile(ctx)
if len(w.a.created) != 1 {
t.Fatalf("created %d times", len(w.a.created))
}
p := w.a.created[0]
if p.As != "mesh_ace_letta" || p.Password != "pw-mesh_ace_letta" || p.Consumer != "ace" {
t.Fatalf("%+v", p)
}
}
func TestAConsumerNoLongerAskedForIsRetiredOnceStable(t *testing.T) {
w := newWorld(t)
w.give(map[string]any{"as": "a"}, map[string]any{"as": "b"})
w.h.Reconcile(ctx)
w.give(map[string]any{"as": "a"})
w.settle() // five passes and ten minutes
if strings.Join(w.a.removed, ",") != "b" {
t.Fatal(w.a.removed)
}
// Only a file that says nobody asks retires the last one.
w.give()
w.settle()
if strings.Join(w.a.removed, ",") != "b,a" {
t.Fatal(w.a.removed)
}
}
func TestNothingReadIsNotNobodyAsking(t *testing.T) {
for name, content := range map[string]string{
"unreadable": "",
"not JSON": "{",
"no given": `{"requirement": "postgres-database"}`,
"another": `{"requirement": "mssql-database", "given": []}`,
} {
t.Run(name, func(t *testing.T) {
w := newWorld(t)
w.give(map[string]any{"as": "a"})
w.h.Reconcile(ctx)
if content == "" {
os.Remove(w.receives)
} else {
os.WriteFile(w.receives, []byte(content), 0o600)
}
w.h.Reconcile(ctx)
if len(w.a.removed) != 0 {
t.Fatalf("retired %v on a file it could not use", w.a.removed)
}
})
}
}
func TestALostConsumerIsAppliedAgainAndBraked(t *testing.T) {
w := newWorld(t)
w.give(map[string]any{"as": "a"})
w.h.Reconcile(ctx)
w.a.held = false
w.now = w.now.Add(2 * time.Minute)
w.h.Reconcile(ctx)
if len(w.a.created) != 2 {
t.Fatalf("created %d times", len(w.a.created))
}
// Still not held a minute later: applied again, then braked for two minutes.
w.now = w.now.Add(61 * time.Second)
w.h.Reconcile(ctx)
w.now = w.now.Add(61 * time.Second)
w.h.Reconcile(ctx)
if len(w.a.created) != 3 {
t.Fatalf("not braked: created %d times", len(w.a.created))
}
}
func TestAFailingCreateIsRetriedAndSaidOnce(t *testing.T) {
w := newWorld(t)
w.a.failing = &pgErr{"extension \"nope\" refused, password pw-a"}
w.give(map[string]any{"as": "a"})
for i := 0; i < 5; i++ {
w.h.Reconcile(ctx)
}
n := 0
for _, s := range w.said {
if strings.Contains(s, "create failed") {
n++
}
}
if n != 1 {
t.Fatalf("said %d times", n)
}
w.a.failing = nil
w.h.Reconcile(ctx)
if len(w.a.created) != 1 {
t.Fatal("not retried")
}
}
type pgErr struct{ s string }
func (e *pgErr) Error() string { return e.s }
func TestScrubRemovesThePassword(t *testing.T) {
if got := scrub(&pgErr{"bad pw a/b c and a%2Fb+c"}, "a/b c"); strings.Contains(got, "a/b c") || strings.Contains(got, "a%2Fb+c") {
t.Fatal(got)
}
}
go/provider/harness/moved.go
+339
View File
@@ -0,0 +1,339 @@
package harness
// A provider handed to the provisioner (novox/hq ADR 0373, ADR 0349 clause 12, issues 780 and 805): where its
// consumers file is set, it runs on the provider helper (package provider) instead of the harness reading the
// controller's grant files. Moved is that path, shared by every provider that keeps a resource per consumer; a module
// says only what differs, in a MovedSpec.
//
// The helper learns who binds the provider from the plans, composes each consumer's entry and reads its credential
// through ConsumersFile, the file this machine's provisioner client fills. It leaves the rest to the two hooks
// here: Make brings the consumer's resource into being exactly as the harness's adapter does, and Gone leaves
// retirement to the harness's guards (retirement.go): the resource is disabled and marked to delete, never removed.
//
// **A consumer not moved is still the grant file's** (issue 937, ADR 0349 clause 17), for a provider whose spec says
// WithGrants: only a moved consumer's plan names its provider, so the controller still gives every other consumer
// through the grant file (MESH_RECEIVES) until R1 (issue 682). Each login it gives is asked for, and one no plan
// names is made and checked by the harness's grant pass exactly as on the grant path; a login a plan names is the
// helper's alone. A grant file that cannot be read is not nobody asking (issue 241): nothing is retired until it is
// read again.
import (
"context"
"encoding/json"
"errors"
"fmt"
"sync"
"time"
"git.novox.be/novox/mesh-sdk/go/provider"
)
// MovedSpec is what differs per provider on the moved path.
type MovedSpec struct {
// Module and Provision are the provider module's name and what it provides, as its manifest says.
Module, Provision string
// Offer is what it serves: its credential form and served values.
Offer provider.Offer
// Login is the identity a consumer is made under at the provider's end (postgres: the entry's User; keycloak: its
// As); an entry with none is not made.
Login func(e provider.Entry) string
// Values are the values the adapter is handed for a consumer; the entry's contribution when not set.
Values func(e provider.Entry) map[string]any
// Noun names what Login is in words ("login", "client id"), Made what is made ("login", "client"), and Backend
// the service in "no longer holds it" ("the backend", "Keycloak").
Noun, Made, Backend string
// WithGrants makes the logins the grant file gives, that no plan names, asked for and made by the harness's grant
// pass (issue 937).
WithGrants bool
}
// Moved is a provider on the provider helper, with the harness's own memory and retirement behind it: what is held,
// what is counted unasked, what waits for a person (retirement.go). The harness is never Run here; the helper's Make
// and Gone and the loop below drive the same state, so the guards are the harness's, not a copy.
type Moved struct {
h *Harness
spec MovedSpec
// FirstRetry is the first wait before a watch asked too early is asked again (issue 943); a second when zero.
FirstRetry time.Duration
mu sync.Mutex // guards assignments and loaded
assignments map[string][]byte
loaded bool
verified map[string]time.Time
}
func newMoved(h *Harness, spec MovedSpec) *Moved {
h.mu.Lock()
h.init()
h.mu.Unlock()
return &Moved{h: h, spec: spec, assignments: map[string][]byte{}, verified: map[string]time.Time{}}
}
// Make brings one consumer's resource into being under its login and the vault's credential, as the harness does:
// at once when it is new or changed, and otherwise only when the backend no longer holds it, which is asked every
// VerifyEvery.
func (m *Moved) Make(ctx context.Context, c provider.Consumer) error {
h, e := m.h, c.Entry
as := m.spec.Login(e)
if as == "" {
return fmt.Errorf("%s has no %s to make", e.Key(), m.spec.Noun)
}
if len(c.Credential) == 0 {
return fmt.Errorf("%s has no credential to make its %s with", e.Key(), m.spec.Made)
}
h.mu.Lock()
defer h.mu.Unlock()
password := string(c.Credential)
values := orEmpty(e.Contributes)
if m.spec.Values != nil {
values = m.spec.Values(e)
}
p := Provision{As: as, Password: password, Values: values, Consumer: e.Consumer}
hash := hashOf(as, password, p.Values, nil)
if was, ok := h.applied[as]; ok && was.hash == hash {
if h.Now().Sub(m.verified[as]) < h.VerifyEvery {
return nil
}
m.verified[as] = h.Now()
hctx, cancel := context.WithTimeout(ctx, h.HoldsTimeout)
held, err := h.Adapter.Holds(hctx, p)
cancel()
if err != nil {
text := scrub(err, password)
h.failed(as, e.Consumer, h.classOf(err, text), text)
return errors.New(text)
}
if held {
h.succeeded(as)
return nil
}
h.say("%s: %s no longer holds it; applying again", as, m.spec.Backend)
}
if err := h.Adapter.Create(ctx, p); err != nil {
text := scrub(err, password)
h.failed(as, e.Consumer, h.classOf(err, text), text)
return errors.New(text)
}
h.succeeded(as)
h.applied[as] = appliedEntry{hash: hash, node: e.Consumer}
m.verified[as] = h.Now()
h.reenabled(as, e.Consumer)
return nil
}
// Gone is the helper asking whether it may forget a consumer that no longer binds. **It never retires**: retiring is
// the harness's rule, the same unasked set over StablePasses passes and StableFor, at most RetireAtOnce at a time
// (retirement.go, ADR 0230), run by retire below. So a consumer missing from one read of the plans loses nothing:
// the helper is told no, and asks again at its next pass, until the login is retired, or asked for again.
func (m *Moved) Gone(_ context.Context, e provider.Entry) error {
h := m.h
as := m.spec.Login(e)
if as == "" {
return nil // refused: nothing was made for it
}
// The plans and the grant file as they are now, not as the last retirement pass read them: with either not
// read, nothing is let go.
want, read := m.asked(h.Node)
h.mu.Lock()
defer h.mu.Unlock()
asked := want[as]
if m.spec.WithGrants {
_, granted, ok := m.granted(want)
read = read && ok
asked = asked || granted[as]
}
_, held := h.applied[as]
_, retired := h.r.retired[as]
switch {
case !read:
case asked:
return nil // another holder of the same login still binds: nothing to retire
case retired, !held && h.r.seeded:
return nil
}
return fmt.Errorf("%s is kept until its retirement is stable (%d passes and %s) and within the bound, or a "+
"person approves it", as, h.StablePasses, h.StableFor)
}
// plan is as much of an assignment as is read to learn who binds the provider.
type plan struct {
Node string `json:"node"`
Module string `json:"module"`
Provisioning *struct {
Provisions []struct {
Provision string `json:"provision"`
Provider string `json:"provider"`
ProviderModule string `json:"provider-module"`
// As is who the consumer is at its provider's end, its slug first (issue 788).
As string `json:"as"`
} `json:"provisions"`
} `json:"provisioning"`
}
// asked is the logins the plans bind the provider for, and whether they could all be read: a plan that is not the
// controller's, or no read yet, is not an answer, so nothing is counted unasked from it (novox/hq issue 241).
func (m *Moved) asked(node string) (map[string]bool, bool) {
m.mu.Lock()
defer m.mu.Unlock()
if !m.loaded || len(m.assignments) == 0 {
return nil, false
}
want := map[string]bool{}
for key, raw := range m.assignments {
var a plan
if err := json.Unmarshal(raw, &a); err != nil || a.Node+"."+a.Module != key {
return nil, false
}
if a.Provisioning == nil {
continue
}
for _, b := range a.Provisioning.Provisions {
if b.Provider == node && b.ProviderModule == m.spec.Module && b.Provision == m.spec.Provision {
as := b.As
if as == "" {
as = provider.ConsumerIdentity(a.Node, a.Module) // a plan that names none
}
want[as] = true
}
}
}
return want, true
}
// granted is what the grant file gives that no plan names (issue 937), and every login it gives; and whether it
// could be read. The caller holds h.mu. With no grant file named there is nothing given, and that is a read.
func (m *Moved) granted(planned map[string]bool) ([]contribution, map[string]bool, bool) {
h := m.h
if h.Receives == "" {
return nil, map[string]bool{}, true
}
given := h.readContributions()
if given == nil {
return nil, nil, false
}
var own []contribution
logins := map[string]bool{}
for _, g := range given {
logins[g.As] = true
if !planned[g.As] {
own = append(own, g)
}
}
return own, logins, true
}
// retire is one pass of the harness's retirement over what the plans bind and the grant file gives now; the logins
// only the grant file gives are made and checked first, as the grant path does.
func (m *Moved) retire(ctx context.Context) {
h := m.h
want, ok := m.asked(h.Node)
h.mu.Lock()
defer h.mu.Unlock()
var given []contribution
if ok && m.spec.WithGrants {
var granted map[string]bool
given, granted, ok = m.granted(want)
for as := range granted {
want[as] = true
}
}
if !ok {
// Not a result: the passes that must agree before anything is retired start again.
h.r.key, h.r.count = "", 0
return
}
h.seed(ctx, want)
h.apply(ctx, given)
h.retireUnasked(ctx, want)
h.r.lastWant = want
for as := range h.failing {
if !want[as] {
delete(h.failing, as)
}
}
for as := range h.trouble {
if _, held := h.applied[as]; !want[as] && !held {
h.recovered(as, "no longer asked for")
}
}
}
// watch keeps the plans as the bus holds them, from the first read on.
func (m *Moved) watch(ctx context.Context, bus provider.Bus) error {
err := bus.Watch(ctx, provider.AssignmentsBucket, func(key string, value []byte) {
m.mu.Lock()
defer m.mu.Unlock()
if value == nil {
delete(m.assignments, key)
} else {
m.assignments[key] = append([]byte(nil), value...)
}
})
if err == nil {
m.mu.Lock()
m.loaded = true
m.mu.Unlock()
}
return err
}
// Run answers the consumers through the helper and retires through the harness's rules, until ctx ends.
//
// The bus is reached through the runtime that launched this bundle, which answers only once Serve runs, and Run starts
// beside Serve: a watch asked too early is asked again, after a wait that doubles from FirstRetry up to a minute,
// until it is answered or ctx ends, so a start that loses the race never leaves the consumers unanswered (issue 943).
func (m *Moved) Run(ctx context.Context, p *provider.Provider, bus provider.Bus) error {
wait := m.FirstRetry
if wait <= 0 {
wait = time.Second
}
for tries := 1; ; tries++ {
err := m.watch(ctx, bus)
if err == nil {
break
}
if tries == 4 || tries%20 == 0 {
m.h.say("the plans cannot be watched yet (%d tries), asking again: %v", tries, err)
}
select {
case <-ctx.Done():
return ctx.Err()
case <-time.After(wait):
}
wait = min(wait*2, time.Minute)
}
go func() {
for {
m.retire(ctx)
select {
case <-ctx.Done():
return
case <-time.After(m.h.Every):
}
}
}()
return p.Run(ctx)
}
// NewMoved is a provider handed to the provisioner: the helper answering its consumers, and the harness (never run)
// whose retirement and tools it shares. consumersFile is where its machine's client fills its consumers. Run the
// result with Moved.Run, passing the provider and the bus returned with it.
func NewMoved(consumersFile string, h *Harness, spec MovedSpec) (*Moved, *provider.Provider, provider.Bus, error) {
if consumersFile == "" {
return nil, nil, nil, errors.New("no consumers' file")
}
run := newMoved(h, spec)
bus := provider.StateBus(spec.Module)
return run, provider.New(provider.Options{
Node: h.Node,
Module: spec.Module,
Offers: map[string]provider.Offer{spec.Provision: spec.Offer},
Bus: bus,
Vault: provider.ConsumersFile{Paths: []string{consumersFile}},
Make: run.Make,
Gone: run.Gone,
Every: h.Every,
Logf: func(f string, a ...any) { h.say(f, a...) },
}), bus, nil
}
go/provider/harness/moved_test.go
+480
View File
@@ -0,0 +1,480 @@
package harness
import (
"context"
"encoding/json"
"fmt"
"os"
"path/filepath"
"reflect"
"sync"
"testing"
"time"
"git.novox.be/novox/mesh-sdk/go/provider"
)
const (
ProvisionName = "postgres-database"
servedPort = 5432
)
// postgresSpec is what a postgres-like provider says: a login per consumer, and the grant file's logins.
var postgresSpec = MovedSpec{Module: "postgres", Provision: ProvisionName,
Offer: provider.Offer{Credential: provider.PerConsumer, Serves: map[string]any{"port": servedPort}},
Login: func(e provider.Entry) string { return e.User },
Noun: "login", Made: "login", Backend: "the backend", WithGrants: true}
// bucketBus is the helper's bus over in-memory buckets.
type bucketBus struct {
mu sync.Mutex
buckets map[string]map[string][]byte
}
func (b *bucketBus) Watch(_ context.Context, bucket string, handle func(string, []byte)) error {
b.mu.Lock()
defer b.mu.Unlock()
for k, v := range b.buckets[bucket] {
handle(k, v)
}
return nil
}
func (b *bucketBus) Put(_ context.Context, bucket, key string, value []byte) error {
b.mu.Lock()
defer b.mu.Unlock()
if b.buckets[bucket] == nil {
b.buckets[bucket] = map[string][]byte{}
}
b.buckets[bucket][key] = append([]byte(nil), value...)
return nil
}
func (b *bucketBus) Delete(_ context.Context, bucket, key string) error {
b.mu.Lock()
defer b.mu.Unlock()
delete(b.buckets[bucket], key)
return nil
}
// bind is one consumer's assignment, whose plan names this machine's postgres as the provider and carries what the
// consumer contributes (issue 776).
func (b *bucketBus) bind(node, module string, contributes map[string]any) {
b.bindNamed("postgres", node, module, contributes, nil)
}
// bindNamed is bind to a provider module, with the names the controller composed for the consumer (issue 792).
func (b *bucketBus) bindNamed(provides, node, module string, contributes map[string]any, names map[string]string) {
plan, _ := json.Marshal(map[string]any{"node": node, "module": module, "provisioning": map[string]any{
"provisions": []any{map[string]any{"provision": ProvisionName, "provider": "novox", "provider-module": provides,
"contributes": contributes, "names": names}}}})
if err := b.Put(context.Background(), provider.AssignmentsBucket, node+"."+module, plan); err != nil {
panic(err)
}
}
// consumersFile writes the file the machine's client fills: one credential per consumer.
func consumersFile(t *testing.T, bound map[[2]string]string) string {
var list []provider.ConsumerCredential
for k, password := range bound {
list = append(list, provider.ConsumerCredential{Consumer: k[0], Module: k[1],
Credential: provider.PairName("novox", ProvisionName, k[0], k[1], ""), Value: password})
}
raw, _ := json.Marshal(list)
at := filepath.Join(t.TempDir(), "consumers.json")
if err := os.WriteFile(at, raw, 0o600); err != nil {
t.Fatal(err)
}
return at
}
// movedWorld is postgres on the helper over an in-memory bus, with a recorder for a server and a clock to move.
type movedWorld struct {
t *testing.T
a *recorder
bus *bucketBus
h *Harness
spec MovedSpec
run *Moved
p *provider.Provider
now time.Time
file string
}
func newMovedWorld(t *testing.T, bound map[[2]string]string) *movedWorld {
return newMovedWorldOf(t, bound, postgresSpec)
}
func newMovedWorldOf(t *testing.T, bound map[[2]string]string, spec MovedSpec) *movedWorld {
w := &movedWorld{t: t, spec: spec, a: &recorder{}, bus: &bucketBus{buckets: map[string]map[string][]byte{}},
now: time.Date(2026, 10, 15, 12, 0, 0, 0, time.UTC)}
w.file = consumersFile(t, bound)
w.h = &Harness{Resource: ProvisionName, Adapter: w.a, Node: "novox", Now: func() time.Time { return w.now },
Log: func(string, ...any) {}}
var bus provider.Bus
w.run, w.p, bus, _ = NewMoved(w.file, w.h, spec)
_ = bus
if err := w.run.watch(ctx, w.bus); err != nil {
t.Fatal(err)
}
return w
}
// pass is one helper pass and one retirement pass, five seconds after the last.
func (w *movedWorld) pass() []provider.Outcome {
w.now = w.now.Add(5 * time.Second)
// the plans as the bus holds them now, as the watch would hand them
w.run.assignments = map[string][]byte{}
for k, v := range w.bus.buckets[provider.AssignmentsBucket] {
w.run.assignments[k] = v
}
// the helper reads its buckets from the in-memory bus
w.p = provider.New(provider.Options{Node: "novox", Module: w.spec.Module, Bus: w.bus,
Vault: provider.ConsumersFile{Paths: []string{w.file}},
Offers: map[string]provider.Offer{ProvisionName: w.spec.Offer},
Make: w.run.Make, Gone: w.run.Gone, Logf: func(string, ...any) {}})
out, err := w.p.Pass(ctx)
if err != nil {
w.t.Fatal(err)
}
w.run.retire(ctx)
return out
}
func bindings(n int) (map[[2]string]string, []string) {
bound := map[[2]string]string{}
var logins []string
for i := 1; i <= n; i++ {
module := fmt.Sprintf("app%d", i)
bound[[2]string{"ace", module}] = "pw-" + module
logins = append(logins, provider.ConsumerIdentity("ace", module))
}
return bound, logins
}
// The consumers' file makes the same calls on the adapter as the equivalent grant files do (the harness).
func TestTheConsumersFileMakesWhatTheGrantFilesMake(t *testing.T) {
letta := provider.ConsumerIdentity("ace", "letta")
plain := provider.ConsumerIdentity("ace", "plain")
// Today: the grant files, through the harness.
g := newWorld(t)
g.give(
map[string]any{"as": letta, "node": "ace", "values": map[string]any{"name": "letta", "extensions": []any{"vector"}}},
map[string]any{"as": plain, "node": "ace", "values": map[string]any{"name": "plain"}},
)
g.h.Reconcile(ctx)
// The consumers' file, through the provider helper.
w := newMovedWorld(t, map[[2]string]string{{"ace", "letta"}: "pw-" + letta, {"ace", "plain"}: "pw-" + plain})
w.bus.bind("ace", "letta", map[string]any{"name": "letta", "extensions": []any{"vector"}})
w.bus.bind("ace", "plain", map[string]any{"name": "plain"})
for _, o := range w.pass() {
if o.Status != provider.Answered {
t.Fatalf("%+v", o)
}
}
byAs := map[string]Provision{}
for _, c := range w.a.created {
byAs[c.As] = c
}
if len(w.a.created) != 2 {
t.Fatalf("created %d, want 2", len(w.a.created))
}
for _, want := range g.a.created {
c := byAs[want.As]
if c.As != want.As || c.Password != want.Password || c.Consumer != want.Consumer ||
!reflect.DeepEqual(orEmpty(c.Values), orEmpty(want.Values)) {
t.Fatalf("made %+v, the grant files make %+v", c, want)
}
}
// Unchanged, and held: nothing is made again, within a minute or after it.
w.a.held = true
w.pass()
w.now = w.now.Add(2 * time.Minute)
w.pass()
if len(w.a.created) != 2 {
t.Fatalf("an unchanged pass created again: %d", len(w.a.created))
}
// No longer held: made again.
w.a.held = false
w.now = w.now.Add(2 * time.Minute)
w.pass()
if len(w.a.created) != 4 {
t.Fatalf("not made again: %d", len(w.a.created))
}
}
// One read without a consumer retires nothing: it takes the harness's stable passes and time, and then the entry
// goes and the login is retired, never dropped.
func TestAConsumerMissingFromOneReadRetiresNothing(t *testing.T) {
bound, logins := bindings(2)
w := newMovedWorld(t, bound)
w.bus.bind("ace", "app1", nil)
w.bus.bind("ace", "app2", nil)
w.pass()
// One read without app2.
saved := w.bus.buckets[provider.AssignmentsBucket]["ace.app2"]
w.bus.Delete(ctx, provider.AssignmentsBucket, "ace.app2")
w.pass()
if len(w.a.removed) != 0 {
t.Fatalf("retired %v on one read", w.a.removed)
}
if _, kept := w.bus.buckets[provider.BucketOf("postgres")]["ace.app2.postgres-database"]; !kept {
t.Fatal("the entry went on one read")
}
// It is back: nothing was lost, and the entry never left the bucket.
w.bus.Put(ctx, provider.AssignmentsBucket, "ace.app2", saved)
for i := 0; i < 8; i++ {
w.pass()
}
w.now = w.now.Add(15 * time.Minute)
w.pass()
if len(w.a.removed) != 0 || len(w.bus.buckets[provider.BucketOf("postgres")]) != 2 {
t.Fatalf("retired %v, entries %d", w.a.removed, len(w.bus.buckets[provider.BucketOf("postgres")]))
}
// Gone for good: not retired before StablePasses and StableFor, then retired and its entry removed.
w.bus.Delete(ctx, provider.AssignmentsBucket, "ace.app2")
for i := 0; i < 6; i++ {
w.pass()
}
if len(w.a.removed) != 0 {
t.Fatalf("retired %v before the time was out", w.a.removed)
}
w.now = w.now.Add(11 * time.Minute)
w.pass()
w.pass()
if !reflect.DeepEqual(w.a.removed, []string{logins[1]}) || len(w.a.deleted) != 0 {
t.Fatalf("retired %v, deleted %v", w.a.removed, w.a.deleted)
}
if _, kept := w.bus.buckets[provider.BucketOf("postgres")]["ace.app2.postgres-database"]; kept {
t.Fatal("the entry stayed after the login was retired")
}
}
// More than three at once waits for a person, as the harness's bound says.
func TestMoreThanTheBoundWaitsForAPerson(t *testing.T) {
bound, _ := bindings(5)
w := newMovedWorld(t, bound)
for i := 1; i <= 5; i++ {
w.bus.bind("ace", fmt.Sprintf("app%d", i), nil)
}
w.pass()
for i := 1; i <= 4; i++ {
w.bus.Delete(ctx, provider.AssignmentsBucket, fmt.Sprintf("ace.app%d", i))
}
for i := 0; i < 6; i++ {
w.pass()
}
w.now = w.now.Add(11 * time.Minute)
w.pass()
w.pass()
if len(w.a.removed) != 0 || w.h.r.waiting == nil {
t.Fatalf("retired %v, waiting %v", w.a.removed, w.h.r.waiting)
}
}
// A plan list that was never read is not nobody asking.
func TestNoPlansReadRetiresNothing(t *testing.T) {
bound, _ := bindings(1)
w := newMovedWorld(t, bound)
w.bus.bind("ace", "app1", nil)
w.pass()
w.bus.Delete(ctx, provider.AssignmentsBucket, "ace.app1")
for i := 0; i < 8; i++ {
w.pass()
}
w.now = w.now.Add(20 * time.Minute)
w.pass()
if len(w.a.removed) != 0 {
t.Fatalf("retired %v with no plans read", w.a.removed)
}
}
// A consumer whose credential is not in the file yet is waited for: nothing is made for it.
func TestAConsumerWithoutACredentialIsWaitedFor(t *testing.T) {
w := newMovedWorld(t, map[[2]string]string{})
w.bus.bind("ace", "letta", map[string]any{"name": "letta"})
outcomes := w.pass()
if len(outcomes) != 1 || outcomes[0].Status != provider.Waiting || len(w.a.created) != 0 {
t.Fatalf("%+v, created %v", outcomes, w.a.created)
}
}
// Where the file is not there, postgres takes the grant directory, as before; where it is, the helper.
// The grant path restarts itself when the consumers file is first filled.
// A consumer whose module has a slug keeps the login its plan names, the slug first (issue 788): the adapter makes it
// under that login, and the retirement counts it as asked for, so a pass retires nothing.
func TestASluggedConsumerKeepsTheLoginItsPlanNames(t *testing.T) {
agent := provider.ConsumerIdentity("ace", "agent")
w := newMovedWorld(t, map[[2]string]string{{"ace", "claude-code"}: "pw-agent"})
plan, _ := json.Marshal(map[string]any{"node": "ace", "module": "claude-code", "provisioning": map[string]any{
"provisions": []any{map[string]any{"provision": ProvisionName, "provider": "novox",
"provider-module": "postgres", "as": agent}}}})
if err := w.bus.Put(ctx, provider.AssignmentsBucket, "ace.claude-code", plan); err != nil {
t.Fatal(err)
}
for _, o := range w.pass() {
if o.Status != provider.Answered {
t.Fatalf("%+v", o)
}
}
if len(w.a.created) != 1 || w.a.created[0].As != agent {
t.Fatalf("made %+v, want one under %s", w.a.created, agent)
}
for i := 0; i < 8; i++ {
w.pass()
}
w.now = w.now.Add(15 * time.Minute)
w.pass()
w.pass()
if len(w.a.removed) != 0 {
t.Fatalf("retired %v: the plan's login was counted unasked", w.a.removed)
}
}
// A provider whose spec names the plan's `as` as the login and no grant file (the keycloak shape) makes the consumer
// under that name with the values its spec composes, and a grant file that cannot be read does not hold up its
// retirement pass.
func TestASpecWithoutGrantsMakesUnderThePlansAsAndIgnoresTheGrantFile(t *testing.T) {
a := &recorder{}
bus := &bucketBus{buckets: map[string]map[string][]byte{}}
now := time.Date(2026, 10, 15, 12, 0, 0, 0, time.UTC)
h := &Harness{Resource: ProvisionName, Adapter: a, Node: "novox", Now: func() time.Time { return now },
Receives: filepath.Join(t.TempDir(), "unreadable-or-missing.json"), Log: func(string, ...any) {}}
spec := MovedSpec{Module: "keycloak", Provision: ProvisionName,
Offer: provider.Offer{Credential: provider.PerConsumer},
Login: func(e provider.Entry) string { return e.As }, Noun: "client id", Made: "client", Backend: "Keycloak",
Values: func(e provider.Entry) map[string]any { return map[string]any{"label": e.Contributes["label"]} }}
file := consumersFile(t, map[[2]string]string{{"ace", "wiki"}: "pw-wiki"})
run, _, _, err := NewMoved(file, h, spec)
if err != nil {
t.Fatal(err)
}
plan, _ := json.Marshal(map[string]any{"node": "ace", "module": "wiki", "provisioning": map[string]any{
"provisions": []any{map[string]any{"provision": ProvisionName, "provider": "novox",
"provider-module": "keycloak", "as": "wiki-client", "contributes": map[string]any{"label": "Wiki"}}}}})
_ = bus.Put(ctx, provider.AssignmentsBucket, "ace.wiki", plan)
if err := run.watch(ctx, bus); err != nil {
t.Fatal(err)
}
p := provider.New(provider.Options{Node: "novox", Module: "keycloak", Bus: bus,
Vault: provider.ConsumersFile{Paths: []string{file}},
Offers: map[string]provider.Offer{ProvisionName: spec.Offer}, Make: run.Make, Gone: run.Gone,
Logf: func(string, ...any) {}})
out, err := p.Pass(ctx)
if err != nil || len(out) != 1 || out[0].Status != provider.Answered {
t.Fatalf("%v %+v", err, out)
}
run.retire(ctx)
if len(a.created) != 1 || a.created[0].As != "wiki-client" || a.created[0].Password != "pw-wiki" ||
a.created[0].Values["label"] != "Wiki" {
t.Fatalf("made %+v", a.created)
}
}
const testIssuer = "https://id.example.org/realms/Novox"
var (
grafanaContributes = map[string]any{"label": "grafana", "endpoint": "web", "callback": "/login/generic_oauth"}
grafanaNames = map[string]string{"name": "grafana.example.org", "internal-name": "grafana.home.internal"}
)
// keycloakSpec is what a keycloak-like provider says: the client id is the plan's `as`, the values are what the
// consumer contributes with the controller's names over them, and no grant file is consulted.
var keycloakSpec = MovedSpec{Module: "keycloak", Provision: ProvisionName,
Offer: provider.Offer{Credential: provider.PerConsumer, Serves: map[string]any{"issuer": testIssuer,
"token-path": "/protocol/openid-connect/token"}},
Login: func(e provider.Entry) string { return e.As },
Values: func(e provider.Entry) map[string]any {
values := map[string]any{}
for k, v := range e.Contributes {
values[k] = v
}
for k, name := range e.Names {
values[k] = name
}
return values
},
Noun: "client id", Made: "client", Backend: "Keycloak"}
// The consumers' file makes the same call on the adapter as the equivalent grant file does (the harness), for a
// provider of the keycloak shape: the same client id, secret and values; and the entry tells the consumer what the
// provider serves.
func TestTheConsumersFileMakesWhatTheGrantFileMakesForTheKeycloakShape(t *testing.T) {
grafana := provider.ConsumerIdentity("home", "grafana")
values := map[string]any{"name": grafanaNames["name"], "internal-name": grafanaNames["internal-name"]}
for k, v := range grafanaContributes {
values[k] = v
}
g := newWorld(t)
g.give(map[string]any{"as": grafana, "node": "home", "values": values})
g.h.Reconcile(ctx)
w := newMovedWorldOf(t, map[[2]string]string{{"home", "grafana"}: "pw-" + grafana}, keycloakSpec)
w.bus.bindNamed("keycloak", "home", "grafana", grafanaContributes, grafanaNames)
for _, o := range w.pass() {
if o.Status != provider.Answered {
t.Fatalf("%+v", o)
}
}
if len(w.a.created) != 1 || len(g.a.created) != 1 {
t.Fatalf("created %d through the helper, %d through the harness, want 1 each", len(w.a.created), len(g.a.created))
}
c, want := w.a.created[0], g.a.created[0]
if c.As != want.As || c.Password != want.Password || c.Consumer != want.Consumer ||
!reflect.DeepEqual(orEmpty(c.Values), orEmpty(want.Values)) {
t.Fatalf("made %+v, the grant file makes %+v", c, want)
}
var entry provider.Entry
if err := json.Unmarshal(w.bus.buckets[provider.BucketOf("keycloak")]["home.grafana."+ProvisionName], &entry); err != nil {
t.Fatal(err)
}
if entry.Values["issuer"] != testIssuer || entry.Values["token-path"] != "/protocol/openid-connect/token" {
t.Fatalf("served %v", entry.Values)
}
w.a.held = true
w.pass()
w.now = w.now.Add(2 * time.Minute)
w.pass()
if len(w.a.created) != 1 {
t.Fatalf("an unchanged pass created again: %d", len(w.a.created))
}
w.a.held = false
w.now = w.now.Add(2 * time.Minute)
w.pass()
if len(w.a.created) != 2 {
t.Fatalf("not made again: %d", len(w.a.created))
}
}
// A consumer whose plan names its client id (a slugged module) keeps it, and the retirement counts it as asked for.
func TestAConsumerNamedByItsPlanKeepsItsClientId(t *testing.T) {
agent := provider.ConsumerIdentity("home", "dashboard")
w := newMovedWorldOf(t, map[[2]string]string{{"home", "grafana-2"}: "pw"}, keycloakSpec)
plan, _ := json.Marshal(map[string]any{"node": "home", "module": "grafana-2", "provisioning": map[string]any{
"provisions": []any{map[string]any{"provision": ProvisionName, "provider": "novox", "provider-module": "keycloak",
"as": agent, "contributes": grafanaContributes, "names": grafanaNames}}}})
if err := w.bus.Put(ctx, provider.AssignmentsBucket, "home.grafana-2", plan); err != nil {
t.Fatal(err)
}
for _, o := range w.pass() {
if o.Status != provider.Answered {
t.Fatalf("%+v", o)
}
}
if len(w.a.created) != 1 || w.a.created[0].As != agent {
t.Fatalf("made %+v, want one under %s", w.a.created, agent)
}
for i := 0; i < 8; i++ {
w.pass()
}
w.now = w.now.Add(15 * time.Minute)
w.pass()
w.pass()
if len(w.a.removed) != 0 {
t.Fatalf("retired %v: the plan's client id was counted unasked", w.a.removed)
}
}
go/provider/harness/retirement.go
+623
View File
@@ -0,0 +1,623 @@
package harness
// What becomes of a consumer the mesh no longer asks for (novox/hq ADR 0230 — the operator's model,
// decided 2026-10-06; it replaces ADR 0229's withdrawal brake, which let one consumer go every hour).
//
// A consumer — a login, a client, a key, and the data behind it — is in one of three states:
//
// 1. ACTIVE.
// 2. RETIRED. The mesh stopped asking for it: its access is disabled, reversibly, and its login and
// data are marked "to delete" with when and why. Nothing is deleted. Asked for again, the ordinary
// create re-enables it at once, as it was.
// 3. DELETED, only through `cleanup delete`, a person's act, which this provider executes because it
// owns its backend.
//
// **Stable removals.** A consumer is retired only once the same set of consumers has gone unasked BOTH
// in StablePasses (5) consecutive passes that read the file AND for at least StableFor (10 minutes)
// since the first pass that saw it. Five passes alone are twenty-five seconds — shorter than a
// controller restart, a store reconnecting or a file half written. A pass that could not read the file
// is not a result and starts both again; so does a different set. Additions and changes are never
// delayed.
//
// **Too many is a person.** A stable set of more than RetireAtOnce (3), or — where more than one is
// held — of more than RetireFraction (half) of those held, retires nothing: it WAITS, said in the
// journal and announced `provisioner.retirement` `waiting` (again every SayAgainEvery), which the
// controller raises as an urgent condition, until `retire approve <node> <module>` or `retire reject`.
// An unassignment the controller made itself is no exception: many changes at once means a person is
// at work, and a person confirms once. On 2026-10-04 one misread file withdrew seven consumers at once
// (issue 241); under this rule that is a question, not an act.
//
// **The backend remembers, not this process.** What is retired, since when and why is read from the
// backend's own mark (Adapter.Inventory), so a restart forgets nothing; a consumer the backend holds
// active that the mesh no longer asks for is retired by the same rules after a restart as before one.
// A consumer found disabled without the mark — withdrawn before this record — is ADOPTED as retired:
// marked, said, announced `adopted`, and its clock starts then.
//
// **A rejection lives as long as this process.** Rejected, the set is kept active and not asked about
// again while it stays the same set; a restart asks again — loudly, never silently.
import (
"context"
"errors"
"fmt"
"sort"
"strings"
"time"
stdio "git.novox.be/novox/mesh-sdk/go"
)
// EventRetirement is the one event a provider says about retirement, its `change` saying what
// happened. The controller derives the permission to emit it for every module that receives
// contributions, as it does the standing events (novox/hq ADR 0224, 0230).
const EventRetirement = "provisioner.retirement"
// The changes EventRetirement carries.
const (
ChangeWaiting = "waiting" // a stable set over the bound waits for a person
ChangeSettled = "settled" // a waiting or rejected set ended without being retired
ChangeApproved = "approved" // a person approved the waiting (or rejected) set
ChangeRejected = "rejected" // a person kept the waiting set active
ChangeRetired = "retired" // consumers disabled and marked to delete
ChangeReenabled = "reenabled" // a retired consumer asked for again, enabled as it was
ChangeDeleted = "deleted" // a retired consumer deleted, by a person
ChangeAdopted = "adopted" // found disabled without the mark, now retired on record
)
// StableFor is the least time the same unasked set must hold before it is retired (novox/hq ADR 0230):
// longer than a controller restart, a store reconnecting or a file half written.
const StableFor = 10 * time.Minute
// KindConsumer is a retired consumer. A backend may list other things set aside for deletion under a
// word of its own (postgres: a database renamed aside).
const KindConsumer = "consumer"
// Retired is one thing a provider holds retired, as its backend's mark says.
type Retired struct {
Consumer string
// Node is the consumer's machine, where the mark records it.
Node string
// RetiredAt is when it was retired; zero for one found disabled without the mesh's mark.
RetiredAt time.Time
Why string
// SizeBytes is what deleting it would free; -1 when the backend cannot say.
SizeBytes int64
Kind string
}
// Inventory is what a backend holds that the mesh made.
type Inventory struct {
Active []string
Retired []Retired
// Sizes is what each active consumer keeps on the backend, in bytes, where the backend can say: what
// the controller tells an empty replacement of a consumer's data by (novox/hq ADR 0233).
Sizes map[string]int64
}
// retirement is the loop's memory of the sets it is counting, waiting on and was told to keep.
type retirement struct {
key string // the unasked set, joined
count int // consecutive passes that saw it
firstSeen time.Time
waiting *waitingSet
rejected *rejectedSet
retired map[string]retiredEntry
lastWant map[string]bool
seeded bool
seedSaid string
// noneSaid is whether this process has said that nothing waits: an earlier process may have left a set
// waiting that this one never knew (issue 937).
noneSaid bool
}
type waitingSet struct {
set []string
since time.Time
saidAt time.Time
held int
}
type rejectedSet struct {
set []string
by, why string
at time.Time
}
type retiredEntry struct {
node string
at time.Time
why string
}
// seed reads what the backend holds, once per process: an active consumer the mesh no longer asks
// for is held, so it is retired by the same rules as one this process made; one found disabled
// without the mark is adopted as retired.
func (h *Harness) seed(ctx context.Context, want map[string]bool) {
if h.r.seeded {
return
}
inv, err := h.Adapter.Inventory(ctx)
if err != nil {
if said := err.Error(); said != h.r.seedSaid {
h.say("cannot list what the backend holds (%v); a consumer the mesh stopped asking for while this "+
"provider was down is not retired until it can", err)
h.r.seedSaid = said
}
return
}
h.r.seeded = true
for _, as := range inv.Active {
if _, held := h.applied[as]; !held && !want[as] {
// No hash: asked for again, it is applied again, which is harmless and marks it.
h.applied[as] = appliedEntry{}
h.say("%s: held by the backend and no longer asked for; retired once that holds for %d passes "+
"and %s (novox/hq ADR 0230)", as, h.StablePasses, h.StableFor)
}
}
now := h.Now()
for _, r := range inv.Retired {
if r.Kind != "" && r.Kind != KindConsumer {
continue
}
if !r.RetiredAt.IsZero() {
h.r.retired[r.Consumer] = retiredEntry{node: r.Node, at: r.RetiredAt, why: r.Why}
continue
}
if want[r.Consumer] {
continue // asked for: this pass's create enables it
}
why := "found disabled without the mesh's mark — withdrawn before retirement was recorded " +
"(novox/hq ADR 0230); retired as found"
if err := h.Adapter.Retire(ctx, r.Consumer, nil, why, now); err != nil {
h.say("%s: found disabled and could not be marked retired, will try at the next start: %v", r.Consumer, err)
continue
}
h.r.retired[r.Consumer] = retiredEntry{node: r.Node, at: now, why: why}
h.say("%s: ADOPTED as retired — %s", r.Consumer, why)
h.announce(EventRetirement, h.retirementBody(ChangeAdopted, []map[string]any{
{"consumer": r.Consumer, "node": r.Node, "retired_at": stamp(now), "why": why, "size_bytes": r.SizeBytes},
}, map[string]any{"why": why}))
}
}
// retireUnasked counts the passes that saw the same set no longer asked for and, once it is stable,
// retires it — or, past the bound, waits for a person.
func (h *Harness) retireUnasked(ctx context.Context, want map[string]bool) {
var unasked []string
for as := range h.applied {
if !want[as] {
unasked = append(unasked, as)
}
}
sort.Strings(unasked)
key := strings.Join(unasked, "\x00")
now := h.Now()
switch {
case len(unasked) == 0:
h.settle("every consumer held is asked for again")
h.r.key, h.r.count = "", 0
return
case key != h.r.key:
if h.r.waiting != nil || h.r.rejected != nil {
h.settle("the set the mesh no longer asks for changed")
}
h.r.key, h.r.count, h.r.firstSeen = key, 1, now
h.say("no longer asked for: %s; retired once the same set holds for %d passes and %s",
strings.Join(unasked, ", "), h.StablePasses, h.StableFor)
default:
h.r.count++
}
if h.r.rejected != nil || h.r.count < h.StablePasses || now.Sub(h.r.firstSeen) < h.StableFor {
return
}
if h.overTheBound(len(unasked), len(h.applied)) {
h.wait(unasked)
return
}
why := fmt.Sprintf("the mesh stopped asking for it: the same result in %d consecutive passes over %s, from %s",
h.r.count, now.Sub(h.r.firstSeen).Round(time.Second), stamp(h.r.firstSeen))
h.retire(ctx, unasked, why, nil)
}
// overTheBound says retiring n of the held consumers at once needs a person.
func (h *Harness) overTheBound(n, held int) bool {
if n == 0 {
return false
}
return n > h.RetireAtOnce || (held > 1 && float64(n) > h.RetireFraction*float64(held))
}
func (h *Harness) bound(held int) string {
return fmt.Sprintf("more than %d at once, or more than %.0f%% of the %d held", h.RetireAtOnce,
h.RetireFraction*100, held)
}
// wait holds a stable set over the bound for a person, said once and announced again every
// SayAgainEvery so a controller that missed the first hears the next.
func (h *Harness) wait(set []string) {
now := h.Now()
w := h.r.waiting
if w == nil {
w = &waitingSet{set: set, since: now, held: len(h.applied)}
h.r.waiting = w
h.say("RETIREMENT WAITS FOR A PERSON: the mesh no longer asks for %d of the %d consumer(s) this provider "+
"holds (%s), %s. Nothing is retired; each stays active until `retire approve %s <module>` or "+
"`retire reject` (novox/hq ADR 0230)", len(set), w.held, strings.Join(set, ", "), h.bound(w.held), h.Node)
} else if now.Sub(w.saidAt) < h.SayAgainEvery {
return
}
w.saidAt = now
h.announce(EventRetirement, h.retirementBody(ChangeWaiting, h.named(set), map[string]any{
"held": w.held, "bound": h.bound(w.held), "since": stamp(w.since),
}))
}
// settle ends a waiting or rejected set that the mesh's own answer made moot.
func (h *Harness) settle(why string) {
var set []string
switch {
case h.r.waiting != nil:
set = h.r.waiting.set
case h.r.rejected != nil:
set = h.r.rejected.set
case h.r.noneSaid:
return
default:
// A restart forgets a set an earlier process waited on, and so would never say it ended: the
// controller's condition would ask a person to approve what nobody waits on any more (issue 937).
// The first time this process finds every consumer held asked for, it says so once, with no set.
h.r.noneSaid = true
h.say("nothing waits for a person: %s", why)
h.announce(EventRetirement, h.retirementBody(ChangeSettled, []map[string]any{}, map[string]any{"why": why}))
return
}
h.r.waiting, h.r.rejected = nil, nil
h.say("retirement of %s settled without retiring: %s", strings.Join(set, ", "), why)
h.announce(EventRetirement, h.retirementBody(ChangeSettled, h.named(set), map[string]any{"why": why}))
}
// retire disables and marks each consumer of set; one that fails stays held and is tried again once
// its set is stable again.
func (h *Harness) retire(ctx context.Context, set []string, why string, extra map[string]any) []string {
now := h.Now()
held := len(h.applied)
var done []string
var said []map[string]any
for _, as := range set {
was, ok := h.applied[as]
if !ok {
continue
}
if err := h.Adapter.Retire(ctx, as, was.derived, why, now); err != nil {
h.say("%s: retiring failed, will try again: %v", as, err)
continue
}
delete(h.applied, as)
delete(h.lost, as)
delete(h.failing, as)
h.r.retired[as] = retiredEntry{node: was.node, at: now, why: why}
h.recovered(as, "retired")
h.say("%s: RETIRED — its access disabled and its data kept, marked to delete (%s). Asked for again "+
"it is re-enabled as it was; it is deleted only by `cleanup delete` (novox/hq ADR 0230)", as, why)
done = append(done, as)
said = append(said, map[string]any{"consumer": as, "node": was.node, "retired_at": stamp(now), "why": why})
}
h.r.key, h.r.count, h.r.waiting, h.r.rejected = "", 0, nil, nil
if len(done) > 0 {
body := map[string]any{"held": held, "why": why}
for k, v := range extra {
body[k] = v
}
h.announce(EventRetirement, h.retirementBody(ChangeRetired, said, body))
}
return done
}
// reenabled says a retired consumer was asked for again and its create enabled it.
func (h *Harness) reenabled(as, node string) {
e, was := h.r.retired[as]
if !was {
return
}
delete(h.r.retired, as)
h.say("%s: asked for again — re-enabled as it was, its mark to delete cleared (retired %s: %s)", as,
stamp(e.at), e.why)
h.announce(EventRetirement, h.retirementBody(ChangeReenabled, []map[string]any{
{"consumer": as, "node": node, "retired_at": stamp(e.at), "why": e.why},
}, nil))
}
// Approve retires exactly the set waiting (or the set rejected) — a person's confirmation.
func (h *Harness) Approve(ctx context.Context, consumers []string, why, by, via string) ([]string, error) {
h.mu.Lock()
defer h.mu.Unlock()
h.init()
if strings.TrimSpace(why) == "" {
return nil, errors.New("approve: say why")
}
set := sorted(consumers)
var held []string
switch {
case h.r.waiting != nil && same(set, h.r.waiting.set):
held = h.r.waiting.set
case h.r.rejected != nil && same(set, h.r.rejected.set):
held = h.r.rejected.set
default:
return nil, fmt.Errorf("approve: %s is not the set waiting here — %s", orNone(set), h.waitingWords())
}
h.say("retirement of %s APPROVED by %s: %s", strings.Join(held, ", "), orSomebody(by), why)
h.announce(EventRetirement, h.retirementBody(ChangeApproved, h.named(held),
map[string]any{"why": why, "by": by, "via": via}))
reason := fmt.Sprintf("the mesh stopped asking for it; approved by %s: %s", orSomebody(by), why)
return h.retire(ctx, held, reason, map[string]any{"by": by, "via": via}), nil
}
// Reject keeps the waiting set active; it is not asked about again while it stays the same set.
func (h *Harness) Reject(consumers []string, why, by, via string) ([]string, error) {
h.mu.Lock()
defer h.mu.Unlock()
h.init()
if strings.TrimSpace(why) == "" {
return nil, errors.New("reject: say why")
}
set := sorted(consumers)
if h.r.waiting == nil || !same(set, h.r.waiting.set) {
return nil, fmt.Errorf("reject: %s is not the set waiting here — %s", orNone(set), h.waitingWords())
}
h.r.rejected = &rejectedSet{set: h.r.waiting.set, by: by, why: why, at: h.Now()}
h.r.waiting = nil
h.say("retirement of %s REJECTED by %s: %s. Kept active, and not asked about again while the mesh goes on "+
"not asking for exactly these (until this provider restarts)", strings.Join(set, ", "), orSomebody(by), why)
h.announce(EventRetirement, h.retirementBody(ChangeRejected, h.named(set),
map[string]any{"why": why, "by": by, "via": via}))
return set, nil
}
// DeleteRetired deletes one retired consumer, by a person's word: never an active one, never one the
// mesh asks for.
func (h *Harness) DeleteRetired(ctx context.Context, consumer, why, by, via string) (int64, error) {
h.mu.Lock()
defer h.mu.Unlock()
h.init()
if strings.TrimSpace(why) == "" {
return 0, errors.New("delete: say why")
}
if h.r.lastWant[consumer] {
return 0, fmt.Errorf("delete: the mesh asks for %s — it is not retired", consumer)
}
if _, held := h.applied[consumer]; held {
return 0, fmt.Errorf("delete: %s is active here — only a retired consumer is deleted", consumer)
}
inv, err := h.Adapter.Inventory(ctx)
if err != nil {
return 0, fmt.Errorf("delete: what the backend holds cannot be read, so nothing is deleted: %w", err)
}
var found *Retired
for i := range inv.Retired {
if inv.Retired[i].Consumer == consumer {
found = &inv.Retired[i]
}
}
if found == nil {
return 0, fmt.Errorf("delete: %s is not retired here — only a retired consumer is deleted", consumer)
}
freed, err := h.Adapter.Delete(ctx, *found)
if err != nil {
return 0, err
}
delete(h.r.retired, consumer)
h.say("%s: DELETED by %s: %s (retired %s: %s; %d bytes freed)", consumer, orSomebody(by), why,
stampOr(found.RetiredAt), found.Why, freed)
h.announce(EventRetirement, h.retirementBody(ChangeDeleted, []map[string]any{{
"consumer": consumer, "node": found.Node, "retired_at": stampOr(found.RetiredAt), "why": found.Why,
"size_bytes": found.SizeBytes, "kind": kindOf(*found),
}}, map[string]any{"why": why, "by": by, "via": via, "freed_bytes": freed}))
return freed, nil
}
// Retirement is what a person (and the controller's `cleanup list` and its probe) asks: what is
// held, waiting, rejected and retired here.
func (h *Harness) Retirement(ctx context.Context) (map[string]any, error) {
h.mu.Lock()
defer h.mu.Unlock()
h.init()
inv, err := h.Adapter.Inventory(ctx)
if err != nil {
return nil, err
}
var held []string
for as := range h.applied {
held = append(held, as)
}
sort.Strings(held)
retired := []map[string]any{}
for _, r := range inv.Retired {
retired = append(retired, map[string]any{"consumer": r.Consumer, "node": r.Node,
"retired_at": stampOr(r.RetiredAt), "why": r.Why, "size_bytes": r.SizeBytes, "kind": kindOf(r)})
}
sizes := map[string]int64{}
for _, as := range held {
if size, ok := inv.Sizes[as]; ok && size >= 0 {
sizes[as] = size
}
}
out := map[string]any{"resource": h.Resource, "node": h.Node, "held": orEmptyList(held), "held_sizes": sizes,
"stable_passes": h.StablePasses, "stable_for_seconds": int(h.StableFor.Seconds()), "bound": h.bound(len(h.applied)), "waiting": nil, "rejected": nil,
"retired": retired}
if w := h.r.waiting; w != nil {
out["waiting"] = map[string]any{"consumers": h.named(w.set), "since": stamp(w.since), "held": w.held}
}
if r := h.r.rejected; r != nil {
out["rejected"] = map[string]any{"consumers": h.named(r.set), "by": r.by, "why": r.why, "at": stamp(r.at)}
}
return out, nil
}
func (h *Harness) waitingWords() string {
switch {
case h.r.waiting != nil:
return "waiting: " + strings.Join(h.r.waiting.set, ", ")
case h.r.rejected != nil:
return "nothing waits; rejected and kept: " + strings.Join(h.r.rejected.set, ", ")
}
return "nothing waits for a person here"
}
// named is a set with each consumer's machine, as the events and the tools say it.
func (h *Harness) named(set []string) []map[string]any {
out := make([]map[string]any, 0, len(set))
for _, as := range set {
out = append(out, map[string]any{"consumer": as, "node": h.applied[as].node})
}
return out
}
func (h *Harness) retirementBody(change string, consumers []map[string]any, extra map[string]any) map[string]any {
body := map[string]any{"provider": h.Resource, "provider-node": h.Node, "change": change,
"consumers": consumers, "at": stamp(h.Now())}
for k, v := range extra {
body[k] = v
}
return body
}
// RetirementTools are the four tools every provider serves for retirement, the same names in every
// module: the controller's `retire` and `cleanup` verbs ask them on the provider's machine.
func RetirementTools(h *Harness) []stdio.Tool {
if h == nil {
return nil
}
ask := func() (context.Context, context.CancelFunc) {
return context.WithTimeout(context.Background(), 2*time.Minute)
}
who := map[string]any{
"why": map[string]any{"type": "string", "description": "why — required, kept in the hand-act log"},
"by": map[string]any{"type": "string", "description": "who decided"},
"via": map[string]any{"type": "string", "description": "mesh-controller when asked through its verbs"},
}
withSet := map[string]any{"consumers": map[string]any{"type": "array", "items": map[string]any{"type": "string"},
"description": "the set, exactly as provisioner_retirement names it"}}
for k, v := range who {
withSet[k] = v
}
withOne := map[string]any{
"consumer": map[string]any{"type": "string", "description": "the retired consumer to delete"},
"confirm": map[string]any{"type": "string", "description": "the consumer's name again, to say this is meant"},
}
for k, v := range who {
withOne[k] = v
}
return []stdio.Tool{
{Name: "provisioner_retirement",
Description: "What this provider holds, what waits for a person to approve its retirement, what was rejected, " +
"and every retired consumer with when, why and its size (novox/hq ADR 0230).",
Run: func(map[string]any) (any, error) {
ctx, cancel := ask()
defer cancel()
return h.Retirement(ctx)
}},
{Name: "provisioner_retire_approve",
Description: "Retire exactly the set waiting for a person (or the set rejected earlier): access disabled, data " +
"kept and marked to delete. Use the controller's `retire approve`, which records the hand act.",
Input: withSet,
Run: func(args map[string]any) (any, error) {
ctx, cancel := ask()
defer cancel()
done, err := h.Approve(ctx, strs(args["consumers"]), argString(args, "why"), argString(args, "by"), argString(args, "via"))
if err != nil {
return nil, err
}
return map[string]any{"retired": orEmptyList(done)}, nil
}},
{Name: "provisioner_retire_reject",
Description: "Keep the set waiting for a person active. Use the controller's `retire reject`.",
Input: withSet,
Run: func(args map[string]any) (any, error) {
kept, err := h.Reject(strs(args["consumers"]), argString(args, "why"), argString(args, "by"), argString(args, "via"))
if err != nil {
return nil, err
}
return map[string]any{"kept": kept}, nil
}},
{Name: "provisioner_delete",
Description: "Delete one RETIRED consumer — its login and its data — for good. Refused for anything active or " +
"asked for. Use the controller's `cleanup delete`, which records the hand act.",
Input: withOne,
Run: func(args map[string]any) (any, error) {
consumer := argString(args, "consumer")
if consumer == "" || argString(args, "confirm") != consumer {
return nil, errors.New("delete: name the consumer, and repeat it in confirm")
}
ctx, cancel := ask()
defer cancel()
freed, err := h.DeleteRetired(ctx, consumer, argString(args, "why"), argString(args, "by"), argString(args, "via"))
if err != nil {
return nil, err
}
return map[string]any{"deleted": consumer, "freed_bytes": freed}, nil
}},
}
}
func strs(v any) []string {
list, _ := v.([]any)
out := []string{}
for _, x := range list {
if s, ok := x.(string); ok && s != "" {
out = append(out, s)
}
}
return out
}
func sorted(list []string) []string {
out := append([]string(nil), list...)
sort.Strings(out)
return out
}
func same(a, b []string) bool {
return strings.Join(sorted(a), "\x00") == strings.Join(sorted(b), "\x00")
}
func orNone(set []string) string {
if len(set) == 0 {
return "an empty set"
}
return strings.Join(set, ", ")
}
func orSomebody(by string) string {
if by == "" {
return "a person"
}
return by
}
func orEmptyList(list []string) []string {
if list == nil {
return []string{}
}
return list
}
func kindOf(r Retired) string {
if r.Kind == "" {
return KindConsumer
}
return r.Kind
}
func stamp(t time.Time) string { return t.UTC().Format(time.RFC3339) }
func stampOr(t time.Time) string {
if t.IsZero() {
return ""
}
return stamp(t)
}
func argString(args map[string]any, key string) string {
s, _ := args[key].(string)
return s
}
go/provider/harness/retirement_test.go
+571
View File
@@ -0,0 +1,571 @@
package harness
// Retirement (novox/hq ADR 0230), as the shared loop does it: a consumer the mesh stops asking for is
// retired only after the same result in five consecutive passes, too many at once wait for a person,
// asked for again it is re-enabled, and only a person deletes. The same file in every Go provider.
import (
"context"
"errors"
"fmt"
"os"
"strings"
"testing"
"time"
)
// recorder is a backend that remembers what it holds, active and retired, as a real one's mark does.
type recorder struct {
created []Provision
removed []string // retired, in order
deleted []string
held bool
failing error
holdsErr error
retErr error
inv Inventory
invErr error
}
func (r *recorder) Create(_ context.Context, p Provision) error {
if r.failing != nil {
return r.failing
}
r.created = append(r.created, p)
r.inv.Retired = withoutRetired(r.inv.Retired, p.As)
if !listHas(r.inv.Active, p.As) {
r.inv.Active = append(r.inv.Active, p.As)
}
return nil
}
func (r *recorder) Retire(_ context.Context, as string, _ map[string]any, why string, at time.Time) error {
if r.retErr != nil {
return r.retErr
}
r.removed = append(r.removed, as)
r.inv.Active = without(r.inv.Active, as)
r.inv.Retired = append(withoutRetired(r.inv.Retired, as),
Retired{Consumer: as, RetiredAt: at, Why: why, SizeBytes: 42, Kind: KindConsumer})
return nil
}
func (r *recorder) Holds(context.Context, Provision) (bool, error) {
if r.holdsErr != nil {
return false, r.holdsErr
}
return r.held, nil
}
func (r *recorder) Inventory(context.Context) (Inventory, error) { return r.inv, r.invErr }
func (r *recorder) Delete(_ context.Context, x Retired) (int64, error) {
r.deleted = append(r.deleted, x.Consumer)
r.inv.Retired = withoutRetired(r.inv.Retired, x.Consumer)
return x.SizeBytes, nil
}
func listHas(list []string, s string) bool {
for _, x := range list {
if x == s {
return true
}
}
return false
}
func without(list []string, s string) []string {
var out []string
for _, x := range list {
if x != s {
out = append(out, x)
}
}
return out
}
func withoutRetired(list []Retired, s string) []Retired {
var out []Retired
for _, x := range list {
if x.Consumer != s {
out = append(out, x)
}
}
return out
}
// holding gives the provider n consumers c1…cn and applies them.
func holding(w *world, n int) []map[string]any {
var given []map[string]any
for i := 1; i <= n; i++ {
given = append(given, map[string]any{"as": fmt.Sprintf("c%d", i), "node": "anchor"})
}
w.give(given...)
w.h.Reconcile(ctx)
return given
}
// pass is one reconcile five seconds after the last.
func (w *world) pass() {
w.now = w.now.Add(5 * time.Second)
w.h.Reconcile(ctx)
}
func retirements(said []announced) []string {
var out []string
for _, a := range said {
if a.event == EventRetirement {
out = append(out, a.body["change"].(string))
}
}
return out
}
func noneWaits(a announced) bool {
c, _ := a.body["consumers"].([]map[string]any)
return a.body["change"] == ChangeSettled && len(c) == 0
}
func lastRetirement(t *testing.T, said []announced) map[string]any {
t.Helper()
for i := len(said) - 1; i >= 0; i-- {
if said[i].event == EventRetirement {
return said[i].body
}
}
t.Fatal("nothing about retirement was announced")
return nil
}
func namesOf(body map[string]any) string {
var out []string
for _, c := range body["consumers"].([]map[string]any) {
out = append(out, c["consumer"].(string))
}
return strings.Join(out, ",")
}
// settle passes every five seconds until the same answer has held for StableFor, and one pass more.
func (w *world) settle() { w.passes(StableFor + 5*time.Second) }
func TestATransientEmptyListForFourPassesRetiresNothing(t *testing.T) {
w, said := standingWorld(t)
given := holding(w, 1)
w.give()
for i := 0; i < 4; i++ {
w.pass()
}
w.give(given...)
w.pass()
if len(w.a.removed) != 0 || len(retirements(*said)) != 0 {
t.Fatalf("four passes retired %v and said %v", w.a.removed, retirements(*said))
}
}
// Five identical passes are twenty-five seconds — shorter than a controller restart. Both must hold:
// five passes AND ten minutes of the same answer (novox/hq ADR 0230).
func TestFivePassesInTwentyFiveSecondsRetireNothingTenMinutesDo(t *testing.T) {
w, said := standingWorld(t)
holding(w, 1)
w.give()
for i := 0; i < 5; i++ {
w.pass()
}
if len(w.a.removed) != 0 || len(retirements(*said)) != 0 {
t.Fatalf("five passes in 25 s retired %v", w.a.removed)
}
w.passes(StableFor - 30*time.Second)
if len(w.a.removed) != 0 {
t.Fatalf("retired before the set held %s: %v", StableFor, w.a.removed)
}
w.passes(time.Minute)
if strings.Join(w.a.removed, ",") != "c1" {
t.Fatalf("the same set held %s and was not retired: %v", StableFor, w.a.removed)
}
// Ten minutes in one slow pass are not five passes.
w2 := newWorld(t)
holding(w2, 1)
w2.give()
w2.pass()
w2.now = w2.now.Add(StableFor)
w2.pass()
if len(w2.a.removed) != 0 {
t.Fatalf("two passes ten minutes apart retired %v", w2.a.removed)
}
w2.passes(15 * time.Second)
if len(w2.a.removed) != 1 {
t.Fatalf("five passes over ten minutes did not retire: %v", w2.a.removed)
}
}
func TestAStableSetIsRetiredAndAskedAgainReEnables(t *testing.T) {
w, said := standingWorld(t)
given := holding(w, 3)
w.give(given[:2]...)
w.settle()
if strings.Join(w.a.removed, ",") != "c3" {
t.Fatalf("retired %v", w.a.removed)
}
body := lastRetirement(t, *said)
if body["change"] != ChangeRetired || namesOf(body) != "c3" || body["held"] != 3 || body["provider-node"] != "anchor" ||
!strings.Contains(body["why"].(string), "consecutive passes over 10m") {
t.Fatalf("%v", body)
}
if len(w.a.inv.Retired) != 1 || w.a.inv.Retired[0].Consumer != "c3" {
t.Fatalf("the backend does not hold it retired: %+v", w.a.inv)
}
// Asked for again: the ordinary create, on the first pass, and said.
created := len(w.a.created)
w.give(given...)
w.pass()
if len(w.a.created) != created+1 || w.a.created[created].As != "c3" {
t.Fatalf("not created again: %v", w.a.created)
}
if body := lastRetirement(t, *said); body["change"] != ChangeReenabled || namesOf(body) != "c3" {
t.Fatalf("%v", body)
}
if len(w.a.inv.Retired) != 0 {
t.Fatalf("still marked retired: %+v", w.a.inv.Retired)
}
}
func TestAnUnreadablePassStartsTheCountAgain(t *testing.T) {
w := newWorld(t)
holding(w, 1)
w.give()
w.passes(StableFor - time.Minute)
os.WriteFile(w.receives, []byte("{"), 0o600)
w.pass()
w.give()
w.passes(StableFor - time.Minute)
if len(w.a.removed) != 0 {
t.Fatalf("an unreadable pass counted: %v", w.a.removed)
}
w.passes(2 * time.Minute)
if len(w.a.removed) != 1 {
t.Fatalf("not retired after ten minutes of readable passes: %v", w.a.removed)
}
}
func TestADifferentSetStartsTheCountAgain(t *testing.T) {
w := newWorld(t)
given := holding(w, 5)
w.give(given[:4]...)
w.passes(StableFor - time.Minute)
w.give(given[:3]...)
w.passes(StableFor - time.Minute)
if len(w.a.removed) != 0 {
t.Fatalf("a changed set kept its count: %v", w.a.removed)
}
w.passes(2 * time.Minute)
if strings.Join(w.a.removed, ",") != "c4,c5" {
t.Fatalf("%v", w.a.removed)
}
}
func TestAdditionsAndChangesAreNeverDelayed(t *testing.T) {
w := newWorld(t)
holding(w, 1)
w.give(map[string]any{"as": "c1", "node": "anchor"}, map[string]any{"as": "c2"})
w.pass()
if len(w.a.created) != 2 || w.a.created[1].As != "c2" {
t.Fatalf("an addition waited: %v", w.a.created)
}
w.give(map[string]any{"as": "c1", "node": "anchor", "values": map[string]any{"name": "rotated"}}, map[string]any{"as": "c2"})
w.pass()
if len(w.a.created) != 3 || w.a.created[2].As != "c1" {
t.Fatalf("a change waited: %v", w.a.created)
}
}
func TestTooManyWaitsForAPersonAndApproveRetiresExactlyThatSet(t *testing.T) {
w, said := standingWorld(t)
holding(w, 4)
w.give()
w.passes(15 * time.Minute)
if len(w.a.removed) != 0 {
t.Fatalf("four of four were retired without a person: %v", w.a.removed)
}
if got := strings.Join(retirements(*said), ","); got != ChangeWaiting {
t.Fatalf("said %q, want one waiting", got)
}
body := lastRetirement(t, *said)
if namesOf(body) != "c1,c2,c3,c4" || body["held"] != 4 || body["bound"] == "" {
t.Fatalf("%v", body)
}
if !strings.Contains(strings.Join(w.said, "\n"), "RETIREMENT WAITS FOR A PERSON") {
t.Fatal("the wait was not said")
}
// Said again every quarter of an hour while it waits.
w.passes(11 * time.Minute)
if got := strings.Join(retirements(*said), ","); got != "waiting,waiting" {
t.Fatalf("%q", got)
}
if _, err := w.h.Approve(ctx, []string{"c1", "c2"}, "tidy", "operator", "mesh-controller"); err == nil {
t.Fatal("approved a set that is not the one waiting")
}
if _, err := w.h.Approve(ctx, []string{"c4", "c3", "c2", "c1"}, "", "operator", ""); err == nil {
t.Fatal("approved without a why")
}
done, err := w.h.Approve(ctx, []string{"c4", "c3", "c2", "c1"}, "the old machine is gone", "operator", "mesh-controller")
if err != nil || strings.Join(done, ",") != "c1,c2,c3,c4" || strings.Join(w.a.removed, ",") != "c1,c2,c3,c4" {
t.Fatal(done, err, w.a.removed)
}
changes := retirements(*said)
if strings.Join(changes[len(changes)-2:], ",") != "approved,retired" {
t.Fatalf("%v", changes)
}
if b := lastRetirement(t, *said); b["by"] != "operator" || b["via"] != "mesh-controller" ||
!strings.Contains(b["why"].(string), "the old machine is gone") {
t.Fatalf("%v", b)
}
// Nothing more waits.
w.passes(time.Minute)
if r, _ := w.h.Retirement(ctx); r["waiting"] != nil || len(r["retired"].([]map[string]any)) != 4 {
t.Fatalf("%v", r)
}
}
func TestRejectedIsKeptAndNotAskedAgainUntilTheSetChanges(t *testing.T) {
w, said := standingWorld(t)
given := holding(w, 4)
w.give()
w.settle()
if _, err := w.h.Reject([]string{"c1"}, "no", "operator", ""); err == nil {
t.Fatal("rejected a set that is not the one waiting")
}
kept, err := w.h.Reject([]string{"c1", "c2", "c3", "c4"}, "a person is moving them", "operator", "mesh-controller")
if err != nil || len(kept) != 4 {
t.Fatal(kept, err)
}
w.passes(time.Hour)
if len(w.a.removed) != 0 {
t.Fatalf("a rejected set was retired: %v", w.a.removed)
}
if got := strings.Join(retirements(*said), ","); got != "waiting,rejected" {
t.Fatalf("asked again after a rejection: %q", got)
}
r, _ := w.h.Retirement(ctx)
if r["rejected"] == nil || r["waiting"] != nil {
t.Fatalf("%v", r)
}
// One of them asked for again: a different answer, so the rejection is settled and counting
// starts over — three of four is over the bound again, and waits again.
w.give(given[0])
w.pass()
if got := strings.Join(retirements(*said), ","); got != "waiting,rejected,settled" {
t.Fatalf("%q", got)
}
w.settle()
if got := strings.Join(retirements(*said), ","); got != "waiting,rejected,settled,waiting" {
t.Fatalf("%q", got)
}
// A rejected set can still be approved later.
w2, _ := standingWorld(t)
holding(w2, 4)
w2.give()
w2.settle()
w2.h.Reject([]string{"c1", "c2", "c3", "c4"}, "wait", "operator", "")
if done, err := w2.h.Approve(ctx, []string{"c1", "c2", "c3", "c4"}, "now", "operator", ""); err != nil || len(done) != 4 {
t.Fatal(done, err)
}
}
func TestAWaitingSetAskedForAgainSettles(t *testing.T) {
w, said := standingWorld(t)
given := holding(w, 4)
w.give()
w.settle()
w.give(given...)
w.pass()
if got := strings.Join(retirements(*said), ","); got != "waiting,settled" || len(w.a.removed) != 0 {
t.Fatalf("%q %v", got, w.a.removed)
}
if _, err := w.h.Approve(ctx, []string{"c1", "c2", "c3", "c4"}, "late", "operator", ""); err == nil {
t.Fatal("approved a set that no longer waits")
}
}
func TestDeleteRemovesOnlyThatRetiredConsumer(t *testing.T) {
w, said := standingWorld(t)
given := holding(w, 5)
w.give(given[:3]...)
w.settle()
if strings.Join(w.a.removed, ",") != "c4,c5" {
t.Fatalf("%v", w.a.removed)
}
if _, err := w.h.DeleteRetired(ctx, "c1", "tidy", "operator", ""); err == nil {
t.Fatal("deleted an active consumer")
}
if _, err := w.h.DeleteRetired(ctx, "c5", "", "operator", ""); err == nil {
t.Fatal("deleted without a why")
}
if _, err := w.h.DeleteRetired(ctx, "nobody", "tidy", "operator", ""); err == nil {
t.Fatal("deleted something not retired")
}
freed, err := w.h.DeleteRetired(ctx, "c5", "the experiment is over", "operator", "mesh-controller")
if err != nil || freed != 42 || strings.Join(w.a.deleted, ",") != "c5" {
t.Fatal(freed, err, w.a.deleted)
}
if b := lastRetirement(t, *said); b["change"] != ChangeDeleted || namesOf(b) != "c5" || b["freed_bytes"] != int64(42) {
t.Fatalf("%v", b)
}
r, _ := w.h.Retirement(ctx)
if left := r["retired"].([]map[string]any); len(left) != 1 || left[0]["consumer"] != "c4" {
t.Fatalf("%v", r)
}
// Retired and asked for again before anybody deleted it: refused, it is the mesh's again.
w.give(given[:4]...)
w.pass()
if _, err := w.h.DeleteRetired(ctx, "c4", "tidy", "operator", ""); err == nil {
t.Fatal("deleted a consumer the mesh asks for")
}
}
func TestTheBound(t *testing.T) {
h := &Harness{}
h.init()
for _, c := range []struct{ n, held int }{{1, 1}, {1, 2}, {1, 3}, {3, 7}, {3, 6}, {2, 5}} {
if h.overTheBound(c.n, c.held) {
t.Errorf("%d of %d waits for a person", c.n, c.held)
}
}
for _, c := range []struct{ n, held int }{{2, 2}, {2, 3}, {4, 7}, {4, 10}, {7, 7}} {
if !h.overTheBound(c.n, c.held) {
t.Errorf("%d of %d is retired without a person", c.n, c.held)
}
}
}
func TestARestartRetiresWhatTheBackendHoldsUnaskedAndAdoptsWhatItFindsDisabled(t *testing.T) {
w, said := standingWorld(t)
w.a.inv = Inventory{Active: []string{"kept", "orphan"}, Retired: []Retired{
{Consumer: "locked-before", SizeBytes: 7, Kind: KindConsumer},
{Consumer: "old_deleted_20261005", RetiredAt: time.Date(2026, 10, 5, 0, 0, 0, 0, time.UTC), Kind: "set-aside-database"},
}}
w.give(map[string]any{"as": "kept"})
w.h.Reconcile(ctx)
if b := lastRetirement(t, *said); b["change"] != ChangeAdopted || namesOf(b) != "locked-before" {
t.Fatalf("%v", b)
}
if strings.Join(w.a.removed, ",") != "locked-before" {
t.Fatalf("adopting did not mark it: %v", w.a.removed)
}
w.settle()
if strings.Join(w.a.removed, ",") != "locked-before,orphan" {
t.Fatalf("an orphan the backend holds was not retired: %v", w.a.removed)
}
}
func TestABackendThatCannotBeListedIsSaidAndAskedAgain(t *testing.T) {
w := newWorld(t)
w.a.invErr = errors.New("connection refused")
holding(w, 1)
if !strings.Contains(strings.Join(w.said, "\n"), "cannot list what the backend holds") {
t.Fatal(w.said)
}
w.a.invErr = nil
w.a.inv = Inventory{Active: []string{"c1", "orphan"}}
w.pass()
w.settle()
if strings.Join(w.a.removed, ",") != "orphan" {
t.Fatalf("%v", w.a.removed)
}
}
func TestTheToolsAnswer(t *testing.T) {
w, _ := standingWorld(t)
holding(w, 4)
w.give()
w.settle()
tools := map[string]func(map[string]any) (any, error){}
for _, tool := range RetirementTools(w.h) {
tools[tool.Name] = tool.Run
}
if len(tools) != 4 {
t.Fatalf("%d tools", len(tools))
}
got, err := tools["provisioner_retirement"](nil)
if err != nil || got.(map[string]any)["waiting"] == nil {
t.Fatal(got, err)
}
set := []any{"c1", "c2", "c3", "c4"}
if _, err := tools["provisioner_retire_approve"](map[string]any{"consumers": set}); err == nil {
t.Fatal("approved without a why")
}
if _, err := tools["provisioner_retire_approve"](map[string]any{"consumers": set, "why": "gone", "by": "operator"}); err != nil {
t.Fatal(err)
}
if _, err := tools["provisioner_delete"](map[string]any{"consumer": "c1", "why": "gone"}); err == nil {
t.Fatal("deleted without confirm")
}
if _, err := tools["provisioner_delete"](map[string]any{"consumer": "c1", "confirm": "c1", "why": "gone"}); err != nil {
t.Fatal(err)
}
if strings.Join(w.a.deleted, ",") != "c1" {
t.Fatal(w.a.deleted)
}
}
func TestAFailingConsumerRetiredIsSaidRecovered(t *testing.T) {
w, said := standingWorld(t)
given := holding(w, 2)
w.a.held = false
w.a.failing = errors.New("boom")
w.passes(7 * time.Minute)
w.give(given[0])
w.settle()
recovered := false
for _, a := range *said {
if a.event == EventRecovered && a.body["consumer"] == "c2" && a.body["why"] == "retired" {
recovered = true
}
}
if !recovered {
t.Fatalf("%v", *said)
}
}
// everyWord is a world that keeps every word the provider announces.
func everyWord(t *testing.T) (*world, *[]announced) {
w := newWorld(t)
var said []announced
w.h.Announce = func(e string, b map[string]any) { said = append(said, announced{e, b}) }
w.h.Node = "anchor"
return w, &said
}
// A restarted provider that finds every consumer it holds asked for says once that nothing waits, so the controller
// clears a waiting condition an earlier process raised and this one never knew (issue 937); it says it no more.
func TestARestartedProviderSaysOnceThatNothingWaits(t *testing.T) {
w, said := everyWord(t)
holding(w, 2)
w.pass()
w.pass()
var words []announced
for _, a := range *said {
if a.event == EventRetirement {
words = append(words, a)
}
}
if len(words) != 1 || !noneWaits(words[0]) || words[0].body["provider-node"] != w.h.Node {
t.Fatalf("said %+v, want one settled with no set", words)
}
}
// A provider whose own set waits does not say that nothing waits.
func TestAProviderThatWaitsDoesNotSayNothingWaits(t *testing.T) {
w, said := everyWord(t)
w.a.inv.Active = []string{"c1", "c2", "c3", "c4", "c5"}
w.give(map[string]any{"as": "c1", "node": "anchor"})
w.settle()
for _, a := range *said {
if a.event == EventRetirement && noneWaits(a) {
t.Fatalf("said nothing waits while %v waited", w.h.r.waiting)
}
}
if w.h.r.waiting == nil {
t.Fatal("nothing waited: the test does not hold the case")
}
}
go/provider/harness/standing_test.go
+193
View File
@@ -0,0 +1,193 @@
package harness
// A provider that keeps failing a consumer says so on the bus (novox/hq ADR 0224): not on the first
// failure, which may be a restart; after FailingAfter of failures with no success between; again
// every SayAgainEvery while it lasts; and recovered on the first success, or when the consumer goes.
import (
"errors"
"os"
"strings"
"testing"
"time"
)
type announced struct {
event string
body map[string]any
}
func standingWorld(t *testing.T) (*world, *[]announced) {
w := newWorld(t)
var said []announced
w.h.Announce = func(e string, b map[string]any) {
// The first success since start, and the first word that nothing waits (issue 937), are their own
// tests'; every other test reads past them.
if c, _ := b["consumers"].([]map[string]any); b["why"] != "first-success" &&
!(e == EventRetirement && b["change"] == ChangeSettled && len(c) == 0) {
said = append(said, announced{e, b})
}
}
w.h.Node = "anchor"
return w, &said
}
// passes reconciles every five seconds for d, as Run would.
func (w *world) passes(d time.Duration) {
for end := w.now.Add(d); w.now.Before(end); w.now = w.now.Add(5 * time.Second) {
w.h.Reconcile(ctx)
}
}
func events(said []announced) string {
var out []string
for _, a := range said {
out = append(out, a.event)
}
return strings.Join(out, ",")
}
func TestAConsumerFailedForMinutesIsAnnouncedNamingItAndTheClass(t *testing.T) {
w, said := standingWorld(t)
w.a.failing = errors.New(`token request failed: 401 {"error":"invalid_grant","error_description":"Invalid user credentials"}`)
w.give(map[string]any{"as": "mesh_home_grafana", "node": "home-server"})
w.passes(4 * time.Minute)
if len(*said) != 0 {
t.Fatalf("announced before FailingAfter: %v", events(*said))
}
w.passes(2 * time.Minute)
if events(*said) != EventFailing {
t.Fatalf("want one %s, got %q", EventFailing, events(*said))
}
b := (*said)[0].body
if b["consumer"] != "mesh_home_grafana" || b["node"] != "home-server" || b["class"] != ClassCredentials ||
b["provider"] != "postgres-database" || b["provider-node"] != "anchor" || b["attempts"].(int) < 60 {
t.Fatalf("%v", b)
}
// Said again while it lasts, not every pass.
w.passes(14 * time.Minute)
if events(*said) != EventFailing {
t.Fatalf("repeated too soon: %q", events(*said))
}
w.passes(2 * time.Minute)
if events(*said) != EventFailing+","+EventFailing {
t.Fatalf("not repeated: %q", events(*said))
}
// The first success takes it back.
w.a.failing = nil
w.passes(5 * time.Second)
if events(*said) != EventFailing+","+EventFailing+","+EventRecovered {
t.Fatalf("no recovery: %q", events(*said))
}
if (*said)[2].body["consumer"] != "mesh_home_grafana" {
t.Fatal((*said)[2].body)
}
}
func TestOneSuccessBetweenFailuresStartsTheRunAgain(t *testing.T) {
w, said := standingWorld(t)
w.a.failing = errors.New("connection refused")
w.give(map[string]any{"as": "a"})
w.passes(4 * time.Minute)
w.a.failing = nil
w.passes(5 * time.Second)
w.give(map[string]any{"as": "a", "values": map[string]any{"name": "changed"}})
w.a.failing = errors.New("connection refused")
w.passes(4 * time.Minute)
if len(*said) != 0 {
t.Fatalf("two runs of four minutes are not one of eight: %q", events(*said))
}
}
// The check that failed for a day on 2026-10-05: clients already made, every minute's check refused
// at the token. A check that cannot be asked is a failure too.
func TestACheckThatKeepsFailingIsAFailureToo(t *testing.T) {
w, said := standingWorld(t)
w.give(map[string]any{"as": "a"})
w.h.Reconcile(ctx)
w.a.holdsErr = errors.New("401 invalid_grant")
w.passes(7 * time.Minute)
if events(*said) != EventFailing || (*said)[0].body["class"] != ClassCredentials {
t.Fatalf("%q %v", events(*said), *said)
}
w.a.holdsErr = nil
w.passes(time.Minute + 5*time.Second)
if events(*said) != EventFailing+","+EventRecovered {
t.Fatalf("%q", events(*said))
}
}
func TestAnUnreadableSecretIsAnnouncedAsSuch(t *testing.T) {
w, said := standingWorld(t)
w.give(map[string]any{"as": "a"})
os.Remove(w.dir + "/a.secret")
w.passes(6 * time.Minute)
if events(*said) != EventFailing || (*said)[0].body["class"] != ClassSecret {
t.Fatalf("%q %v", events(*said), *said)
}
}
func TestAConsumerNoLongerAskedForIsNoLongerFailing(t *testing.T) {
w, said := standingWorld(t)
w.a.failing = errors.New("boom")
w.give(map[string]any{"as": "a"}, map[string]any{"as": "b"})
w.passes(6 * time.Minute)
if events(*said) != EventFailing+","+EventFailing {
t.Fatalf("%q", events(*said))
}
w.give(map[string]any{"as": "a"})
w.passes(5 * time.Second)
last := (*said)[len(*said)-1]
if last.event != EventRecovered || last.body["consumer"] != "b" || last.body["why"] != "no longer asked for" {
t.Fatalf("%v", *said)
}
}
func TestAnErrorIsClassedByItsWords(t *testing.T) {
for text, want := range map[string]string{
`Keycloak token request failed: 401 {"error":"invalid_grant"}`: ClassCredentials,
`FATAL: password authentication failed for user "postgres"`: ClassCredentials,
`dial tcp 127.0.0.1:5432: connect: connection refused`: ClassUnreachable,
`context deadline exceeded`: ClassUnreachable,
`extension "nope" is not available`: ClassRefused,
} {
if got := ClassOf(text); got != want {
t.Errorf("%s: %s, want %s", text, got, want)
}
}
}
func TestAnAdapterThatClassesItsOwnErrorsIsBelieved(t *testing.T) {
w, said := standingWorld(t)
w.h.Adapter = classing{w.a}
w.a.failing = errors.New("anything")
w.give(map[string]any{"as": "a"})
w.passes(6 * time.Minute)
if (*said)[0].body["class"] != "its-own" {
t.Fatal((*said)[0].body)
}
}
type classing struct{ *recorder }
func (classing) Class(error) string { return "its-own" }
// A provider restarted after announcing a failure has forgotten it; its first success for each
// consumer is announced, so the controller clears what it kept rather than naming it for ever.
func TestTheFirstSuccessSinceStartIsAnnouncedOnce(t *testing.T) {
w := newWorld(t)
var said []announced
w.h.Announce = func(e string, b map[string]any) {
if e != EventRetirement {
said = append(said, announced{e, b})
}
}
w.give(map[string]any{"as": "a"})
w.passes(3 * time.Minute)
if events(said) != EventRecovered || said[0].body["why"] != "first-success" || said[0].body["consumer"] != "a" {
t.Fatalf("%v", said)
}
}
go/provider/provider.go
+136 -25
View File
@@ -1,5 +1,6 @@
// 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).
// 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
@@ -35,7 +36,9 @@ package provider
import (
"context"
"encoding/base64"
"encoding/json"
"errors"
"fmt"
"regexp"
"sort"
@@ -63,6 +66,15 @@ type Bus interface {
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)
@@ -85,7 +97,8 @@ 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 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
@@ -105,8 +118,10 @@ type Entry struct {
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"`
// 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"`
// Contributes is what the consumer's module tells its provider about this provision (a database's name, its
@@ -119,6 +134,10 @@ type Entry struct {
// 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
@@ -195,6 +214,9 @@ const (
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.
@@ -250,8 +272,13 @@ func (p *Provider) watch(ctx context.Context, bucket string, into map[string][]b
}
// 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.
// 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()
@@ -297,17 +324,25 @@ type planned struct {
From string `json:"from"`
As string `json:"as"`
Names map[string]string `json:"names"`
CredentialForm string `json:"credential-form"`
Name string `json:"name"`
} `json:"provisions"`
} `json:"provisioning"`
}
func (p *Provider) pass(ctx context.Context, assignments, entries map[string][]byte) ([]Outcome, error) {
var binds []Entry
var out []Outcome
unknown := map[string]bool{} // consumers whose assignment was skipped: their entries are not judged
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)
o := Outcome{Key: key, Status: Failed, Why: fmt.Sprintf("the assignment %s is not the controller's, 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] = true
continue
}
if a.Provisioning == nil {
continue
@@ -318,11 +353,11 @@ func (p *Provider) pass(ctx context.Context, assignments, entries map[string][]b
}
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})
Contributes: b.Contributes, At: b.At, From: b.From, As: b.As, Names: b.Names,
CredentialForm: b.CredentialForm, Name: b.Name})
}
}
var out []Outcome
want := map[string]bool{}
for _, b := range binds {
key := b.Key()
@@ -332,7 +367,7 @@ func (p *Provider) pass(ctx context.Context, assignments, entries map[string][]b
p.say(key, o.Status+" "+o.Why)
}
for _, key := range sortedKeys(entries) {
if want[key] {
if want[key] || unknown[consumerOf(key)] {
continue
}
var e Entry
@@ -349,7 +384,14 @@ func (p *Provider) pass(ctx context.Context, assignments, entries map[string][]b
continue
}
}
if err := p.opts.Bus.Delete(ctx, BucketOf(p.opts.Module), key); err != nil {
if err := p.remove(ctx, key, entries[key]); err != nil {
if errors.Is(err, stdio.ErrChanged) {
o := Outcome{Key: key, Status: Superseded, Why: "another instance of the provider wrote its entry " +
"since this pass read it, so it stands"}
out = append(out, o)
p.say(key, o.Status+" "+o.Why)
continue
}
o := Outcome{Key: key, Status: Failed, Why: "no longer bound, and its entry could not be removed: " +
err.Error()}
out = append(out, o)
@@ -365,6 +407,23 @@ func (p *Provider) pass(ctx context.Context, assignments, entries map[string][]b
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()
@@ -416,32 +475,59 @@ func (p *Provider) answer(ctx context.Context, e Entry, have []byte) Outcome {
}
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)
// 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]
}
if e.Credential != "" {
credential, err = p.opts.Vault.Value(ctx, e.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.Credential, err)}
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}); err != nil {
why := err.Error()
if len(credential) > 0 {
why = strings.ReplaceAll(why, string(credential), "<credential>")
}
why = redact(why, 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})
}
// 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 {
@@ -470,6 +556,11 @@ func (p *Provider) say(key, what string) {
// 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 }
// 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 "provision/" + provider + "/" + provision + "/" + node + "/" + 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 {
n := "pair/" + provider + "/" + provision + "/" + node + "/" + module
@@ -626,6 +717,26 @@ func (b stateBus) Put(_ context.Context, bucket, key string, value []byte) error
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 {
go/provider/provider_test.go
+233 -20
View File
@@ -3,6 +3,7 @@ package provider
import (
"bytes"
"context"
"encoding/base64"
"encoding/json"
"errors"
"fmt"
@@ -12,6 +13,8 @@ import (
"sync"
"testing"
"time"
stdio "git.novox.be/novox/mesh-sdk/go"
)
// standIn is a stand-in bus: buckets of keys and values, and watches that hand the current values first and then
@@ -65,6 +68,17 @@ func (s *standIn) Delete(_ context.Context, bucket, key string) error {
return nil
}
// DeleteIf removes the key only while it still holds what the caller read.
func (s *standIn) DeleteIf(ctx context.Context, bucket, key string, held []byte) error {
s.mu.Lock()
cur := s.buckets[bucket][key]
s.mu.Unlock()
if !bytes.Equal(cur, held) {
return fmt.Errorf("%s: %w", key, stdio.ErrChanged)
}
return s.Delete(ctx, bucket, key)
}
func (s *standIn) get(bucket, key string) []byte {
s.mu.Lock()
defer s.mu.Unlock()
@@ -224,7 +238,7 @@ func TestEachConsumerGetsAnEntryOfItsOwnAndTheVaultsValue(t *testing.T) {
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 != "" {
e.VaultName != "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" ||
@@ -263,7 +277,7 @@ func TestASharedCredentialIsTheVaultsOneValueAndTheProvidersOneUser(t *testing.T
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 {
if e.VaultName != "provision/novox/sonarr-api" || e.User != "sonarr" || e.Values["url"] == nil {
t.Fatalf("%s: %+v", key, e)
}
if m.made[key] != "one-value" {
@@ -291,7 +305,7 @@ func TestAMultiHolderConsumerIsRefusedADerivedValueNamingBothEnds(t *testing.T)
t.Fatalf("the refusal does not name %q: %s", says, e.Refused)
}
}
if e.Values != nil || e.Credential != "" {
if e.Values != nil || e.VaultName != "" {
t.Fatalf("a refused entry carries values: %+v", e)
}
if m.calls != 0 {
@@ -317,7 +331,7 @@ func TestALocalNameIsServedWhenNothingIsDerived(t *testing.T) {
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" {
if e.Refused != "" || e.VaultName != "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" {
@@ -404,22 +418,42 @@ func TestOnlyBindsNamingThisProviderAreAnswered(t *testing.T) {
}
}
// 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 assignment that cannot be read is skipped and said, by its key and why; the rest of the pass goes on (issue 825,
// finding 3). Its consumer's own entry is left alone, since a bind missed would read as an unbind.
func TestAnUnreadableAssignmentIsSkippedAndSaidAndTheRestGoesOn(t *testing.T) {
for name, bad := range map[string]string{"not json": "{not json",
"another key": `{"node":"g14","module":"umami"}`} {
bus, m := newStandIn(), &maker{}
v := &vault{values: map[string]string{"pair/novox/postgres-database/ace/letta": "s",
"pair/novox/postgres-database/g14/umami": "u"}}
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(bad))
assign(t, bus, "g14", "umami", bind("postgres-database", "", "novox", "postgres"))
var logged []string
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.Fatalf("%s: %v", name, err)
}
if bus.get(BucketOf("postgres"), "g14.umami.postgres-database") == nil {
t.Fatalf("%s: the good consumer was not answered", name)
}
if bus.get(BucketOf("postgres"), "ace.letta.postgres-database") == nil || len(m.gone) != 0 {
t.Fatalf("%s: the entry of the unreadable assignment's consumer was removed", name)
}
said := false
for _, o := range out {
if o.Key == "ace.letta" && o.Status == Failed && o.Why != "" {
said = true
}
}
if !said || !strings.Contains(strings.Join(logged, "\n"), "ace.letta") {
t.Fatalf("%s: the skipped assignment is not said: %+v, %q", name, out, logged)
}
}
}
@@ -588,3 +622,182 @@ func TestAConsumersEntryCarriesTheNamesOfItsPlan(t *testing.T) {
t.Fatalf("an entry of a plan without names changed: %+v", plain)
}
}
// refusedByTheStateGuard mirrors the key rule of the runtime's state guard (mesh-tools node-tools internal/bus
// state.go): a JSON key whose lowercased name, with `-`, `_` and spaces taken out, ends in one of these is refused.
func refusedByTheStateGuard(key string) bool {
k := strings.NewReplacer("-", "", "_", "", " ", "").Replace(strings.ToLower(key))
for _, end := range []string{"password", "passwd", "secret", "token", "credential", "credentials",
"authorization", "apikey", "privatekey", "accesskey", "cookie"} {
if strings.HasSuffix(k, end) {
return true
}
}
return false
}
func keysOf(v any, into *[]string) {
switch x := v.(type) {
case map[string]any:
for k, c := range x {
*into = append(*into, k)
keysOf(c, into)
}
case []any:
for _, c := range x {
keysOf(c, into)
}
}
}
// Every entry the helper writes passes the state guard's key rule (issue 945), and names where the vault keeps the
// credential under `vault-name`, for a login per consumer and for a shared one.
func TestAnEntryHasNoKeyTheStateGuardRefuses(t *testing.T) {
for name, form := range map[string]CredentialForm{"pair": PerConsumer, "shared": Shared} {
bus, m := newStandIn(), &maker{}
v := &vault{values: map[string]string{"pair/novox/postgres-database/ace/letta": "s1",
"provision/novox/postgres-database": "s2"}}
assign(t, bus, "ace", "letta", bind("postgres-database", "", "novox", "postgres"))
p := New(Options{Node: "novox", Module: "postgres", Bus: bus, Vault: v, Make: m.Make, Logf: func(string, ...any) {},
Offers: map[string]Offer{"postgres-database": {Credential: form, User: "shared-user"}}})
if _, err := p.Pass(context.Background()); err != nil {
t.Fatal(err)
}
var doc map[string]any
if err := json.Unmarshal(bus.get(BucketOf("postgres"), "ace.letta.postgres-database"), &doc); err != nil {
t.Fatal(err)
}
var keys []string
keysOf(doc, &keys)
for _, k := range keys {
if refusedByTheStateGuard(k) {
t.Fatalf("%s: the entry carries the key %q, which the state guard refuses", name, k)
}
}
want := map[string]string{"pair": "pair/novox/postgres-database/ace/letta",
"shared": "provision/novox/postgres-database"}[name]
if doc["vault-name"] != want {
t.Fatalf("%s: vault-name is %v, want %s", name, doc["vault-name"], want)
}
}
}
// The plan's credential form and holder name win over the offer's; the offer's form is the default (issue 825,
// finding 4).
func TestThePlansCredentialFormWinsOverTheOffers(t *testing.T) {
run := func(offer CredentialForm, plan map[string]string) Entry {
bus, m := newStandIn(), &maker{}
v := &vault{values: map[string]string{"pair/novox/postgres-database/ace/letta": "p",
"provision/novox/postgres-database": "s", "provision/novox/postgres-database/ace/letta": "h"}}
b := bind("postgres-database", "", "novox", "postgres")
for k, val := range plan {
b[k] = val
}
assign(t, bus, "ace", "letta", b)
p := New(Options{Node: "novox", Module: "postgres", Bus: bus, Vault: v, Make: m.Make, Logf: func(string, ...any) {},
Offers: map[string]Offer{"postgres-database": {Credential: offer, User: "shared-user"}}})
out, err := p.Pass(context.Background())
if err != nil || len(out) != 1 || out[0].Status != Answered {
t.Fatalf("%v %+v", err, out)
}
return entryOf(t, bus, "ace.letta.postgres-database")
}
if e := run(PerConsumer, map[string]string{"credential-form": "shared"}); e.VaultName !=
"provision/novox/postgres-database" || e.User != "shared-user" {
t.Fatalf("shared over pair: %+v", e)
}
if e := run(Shared, map[string]string{"credential-form": "pair"}); e.VaultName !=
"pair/novox/postgres-database/ace/letta" || e.User != "mesh_ace_letta" {
t.Fatalf("pair over shared: %+v", e)
}
for _, form := range []string{"issued", "given"} {
if e := run(PerConsumer, map[string]string{"credential-form": form, "name": "main"}); e.VaultName !=
"provision/novox/postgres-database/ace/letta" || e.User != "mesh_ace_letta" {
t.Fatalf("%s holder: %+v", form, e)
}
}
if e := run(Shared, nil); e.VaultName != "provision/novox/postgres-database" {
t.Fatalf("the offer's form is the default: %+v", e)
}
}
// An instance deletes only an entry it wrote, and only while the entry is as it read it: one that another instance of
// the provider wrote after the consumer moved stands (issue 825, finding 5).
func TestAnEntryAnotherInstanceWroteAfterTheMoveStands(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"))
a := postgres(bus, v, m)
if _, err := a.Pass(context.Background()); err != nil {
t.Fatal(err)
}
// The consumer moves to the instance on shanks: A's pass reads its own entry and the unbind, and while it lets
// the consumer go, B writes the consumer's entry.
_ = bus.Delete(context.Background(), AssignmentsBucket, "ace.letta")
theirs, _ := json.Marshal(Entry{Provider: "shanks", ProviderModule: "postgres", Provision: "postgres-database",
Consumer: "ace", Module: "letta", VaultName: "pair/shanks/postgres-database/ace/letta"})
a.opts.Gone = func(ctx context.Context, e Entry) error {
return bus.Put(ctx, BucketOf("postgres"), e.Key(), theirs)
}
out, err := a.Pass(context.Background())
if err != nil {
t.Fatal(err)
}
if got := bus.get(BucketOf("postgres"), "ace.letta.postgres-database"); !bytes.Equal(got, theirs) {
t.Fatalf("the other instance's entry was removed: %s", got)
}
for _, o := range out {
if o.Status == Removed {
t.Fatalf("outcomes %+v", out)
}
}
}
// A failed Make is said with the credential taken out in the forms it can come back in: as it is, in base64 (standard
// and URL, with and without padding) and as a JSON string says it (issue 825, finding 6).
func TestAFailedMakeIsSaidWithoutTheCredentialInAnyOfItsForms(t *testing.T) {
cred := "pa/ss+w\"rd?>>x"
b := []byte(cred)
jsonQuoted, _ := json.Marshal(cred)
forms := map[string]string{
"raw": cred, "std": base64.StdEncoding.EncodeToString(b), "std unpadded": base64.RawStdEncoding.EncodeToString(b),
"url": base64.URLEncoding.EncodeToString(b), "url unpadded": base64.RawURLEncoding.EncodeToString(b),
"json": string(jsonQuoted[1 : len(jsonQuoted)-1]),
}
if forms["std"] == forms["url"] || forms["json"] == forms["raw"] {
t.Fatal("the test credential does not tell the forms apart")
}
for name, form := range forms {
bus, m := newStandIn(), &maker{fail: errors.New("login failed for " + form + " at the server")}
v := &vault{values: map[string]string{"pair/novox/postgres-database/ace/letta": cred}}
assign(t, bus, "ace", "letta", bind("postgres-database", "", "novox", "postgres"))
out, err := postgres(bus, v, m).Pass(context.Background())
if err != nil || len(out) != 1 {
t.Fatalf("%v %+v", err, out)
}
if strings.Contains(out[0].Why, form) || !strings.Contains(out[0].Why, "<credential>") {
t.Fatalf("%s form: %q", name, out[0].Why)
}
}
}
// An offer that declares no credential is answered with none whatever form the plan says (the controller writes
// "pair" for such an offer): no vault name, and the Vault is never asked, so a provider with no Vault works (issue 825,
// finding 4; issue 923).
func TestAnOfferWithNoCredentialIgnoresThePlansForm(t *testing.T) {
bus := newStandIn()
b := bind("route", "", "novox", "route-proxy")
b["credential-form"] = "pair"
assign(t, bus, "ace", "letta", b)
p := New(Options{Node: "novox", Module: "route-proxy", Bus: bus, Logf: func(string, ...any) {},
Offers: map[string]Offer{"route": {Credential: NoCredential}}})
out, err := p.Pass(context.Background())
if err != nil || len(out) != 1 || out[0].Status != Answered {
t.Fatalf("%v %+v", err, out)
}
var e Entry
if err := json.Unmarshal(bus.get(BucketOf("route-proxy"), "ace.letta.route"), &e); err != nil || e.VaultName != "" ||
e.User != "" {
t.Fatalf("%v %+v", err, e)
}
}
go/provision.go
+1 -1
View File
@@ -6,7 +6,7 @@ import (
"time"
)
// A module's code reads its values at runtime (novox/hq ADR 0349 clause 14, ADR 0370, design 57 S1): its config
// A module's code reads its values at runtime (novox/hq ADR 0349 clause 14, ADR 0370, and the part of the feature 57 design that lets a running module read, rather than be handed in a file, its provisioned values): its config
// values and its bindings' values that are not secret, for what its manifest declares, and watches them. The
// runtime that launched this bundle asks its machine's provisioner client **as this bundle's module** — the module it
// launched, never a name this code passes — and the client answers only what that module declares, only while the