Bind every provider by mesh-provisioner's plans and provision keys instead of the controller's assignments, act once a provision's dependencies are provided, say the outcome in its entry, and retire a consumer only on the operator's yes (issue 1954 L6-P3, P4, ADR 0570 clauses 2, 3, 5)
This commit is contained in:
@@ -39,8 +39,7 @@ func TestTheConsumersFileAnswersEachConsumersCredential(t *testing.T) {
|
||||
|
||||
// Wired as the helper's Vault, the file is what makes each consumer's login.
|
||||
bus := newStandIn()
|
||||
bus.Put(context.Background(), AssignmentsBucket, "ace.letta", []byte(`{"node":"ace","module":"letta",
|
||||
"provisioning":{"provisions":[{"provision":"postgres-database","provider":"novox","provider-module":"postgres"}]}}`))
|
||||
assign(t, bus, "ace", "letta", bind("postgres-database", "", "novox", "postgres"))
|
||||
var made []string
|
||||
p := New(Options{Node: "novox", Module: "postgres", Bus: bus, Vault: v,
|
||||
Offers: map[string]Offer{"postgres-database": {Serves: map[string]any{"port": 5432}}},
|
||||
|
||||
@@ -1,51 +1,9 @@
|
||||
package provider
|
||||
|
||||
import (
|
||||
"context"
|
||||
"encoding/json"
|
||||
"slices"
|
||||
"testing"
|
||||
)
|
||||
|
||||
// Issue 1098: an assignment with `ended` set, or with a `state`, is not a standing assignment: the helper reads it
|
||||
// as a removed key, so the entry of its consumer is removed and the adapter hears it is gone. The fields the
|
||||
// controller adds change nothing.
|
||||
func TestAnEndedAssignmentIsReadAsARemovedKey(t *testing.T) {
|
||||
bus, m := newStandIn(), &maker{}
|
||||
v := &vault{values: map[string]string{"pair/novox/postgres-database/ace/letta": "s1"}}
|
||||
assign(t, bus, "ace", "letta", bind("postgres-database", "", "novox", "postgres"))
|
||||
p := postgres(bus, v, m)
|
||||
if _, err := p.Pass(context.Background()); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if got := bus.keys(BucketOf("postgres")); !slices.Equal(got, []string{"ace.letta.postgres-database"}) {
|
||||
t.Fatalf("entries %v", got)
|
||||
}
|
||||
for _, mark := range []map[string]any{{"ended": "2026-10-18T01:00:00Z"}, {"state": "pending"}} {
|
||||
raw := bus.get(AssignmentsBucket, "ace.letta")
|
||||
var a map[string]any
|
||||
if err := json.Unmarshal(raw, &a); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
for k, v := range mark {
|
||||
a[k] = v
|
||||
}
|
||||
raw, _ = json.Marshal(a)
|
||||
_ = bus.Put(context.Background(), AssignmentsBucket, "ace.letta", raw)
|
||||
out, err := p.Pass(context.Background())
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if got := bus.keys(BucketOf("postgres")); len(got) != 0 || len(out) != 1 || out[0].Status != Removed {
|
||||
t.Fatalf("%v: entries %v, outcomes %+v", mark, got, out)
|
||||
}
|
||||
assign(t, bus, "ace", "letta", bind("postgres-database", "", "novox", "postgres"))
|
||||
if _, err := p.Pass(context.Background()); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
func TestStandingIsFalseOnlyForAnEndedOrPendingValue(t *testing.T) {
|
||||
for raw, want := range map[string]bool{
|
||||
`{"node":"a","module":"b"}`: true,
|
||||
|
||||
@@ -105,7 +105,8 @@ func TestAnUnreadableGrantFileRetiresNothing(t *testing.T) {
|
||||
}
|
||||
}
|
||||
|
||||
// A login the grant file stops giving is retired by the usual rules: stable, then retired, never dropped.
|
||||
// A login the grant file stops giving is not retired by the provider itself (novox/hq ADR 0570 clause 5): every
|
||||
// retirement waits for the operator's yes on a provision's key.
|
||||
func TestALoginTheGrantFileStopsGivingIsRetired(t *testing.T) {
|
||||
bound, _ := bindings(1)
|
||||
w := newMovedWorld(t, bound)
|
||||
@@ -117,7 +118,7 @@ func TestALoginTheGrantFileStopsGivingIsRetired(t *testing.T) {
|
||||
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)
|
||||
if len(w.a.removed) != 0 || len(w.a.deleted) != 0 {
|
||||
t.Fatalf("retired %v, deleted %v, by itself", w.a.removed, w.a.deleted)
|
||||
}
|
||||
}
|
||||
|
||||
@@ -19,13 +19,13 @@ package harness
|
||||
|
||||
import (
|
||||
"context"
|
||||
"encoding/json"
|
||||
"errors"
|
||||
"fmt"
|
||||
"sync"
|
||||
"time"
|
||||
|
||||
"git.novox.be/novox/mesh-sdk/go/provider"
|
||||
"git.novox.be/novox/mesh-sdk/go/provisions"
|
||||
)
|
||||
|
||||
// MovedSpec is what differs per provider on the moved path.
|
||||
@@ -57,17 +57,17 @@ type Moved struct {
|
||||
// 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
|
||||
mu sync.Mutex // guards plans and loaded
|
||||
plans 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{}}
|
||||
return &Moved{h: h, spec: spec, plans: 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:
|
||||
@@ -122,77 +122,50 @@ func (m *Moved) Make(ctx context.Context, c provider.Consumer) error {
|
||||
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 {
|
||||
// Gone is the helper letting a consumer go, which it does only on a retiring provision the operator said yes to (novox/hq
|
||||
// ADR 0570 clause 5): its login is retired at once, through the adapter, its resource and data kept and marked to
|
||||
// delete, deleted only by a person's `cleanup delete`. A consumer another plan still binds under the same login loses
|
||||
// nothing.
|
||||
func (m *Moved) Gone(ctx 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)
|
||||
want, _ := 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:
|
||||
if want[as] {
|
||||
return nil // another holder of the same login still binds: nothing to retire
|
||||
case retired, !held && h.r.seeded:
|
||||
}
|
||||
if _, held := h.applied[as]; !held {
|
||||
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)
|
||||
if done := h.retire(ctx, []string{as}, "the operator approved its retirement ("+e.Key()+")", nil); len(done) != 1 {
|
||||
return fmt.Errorf("%s was not retired; it is tried again at the next pass", as)
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
// 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).
|
||||
// asked is the logins the seat's plans bind the provider for (novox/hq ADR 0570 clause 2), and whether they could all be
|
||||
// read: a plan that cannot be read, or no read yet, is not an answer (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 {
|
||||
if !m.loaded || len(m.plans) == 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 {
|
||||
for key, raw := range m.plans {
|
||||
pl, err := provisions.DecodePlan(raw)
|
||||
if err != nil || pl.Node+"."+pl.Module != key {
|
||||
return nil, false
|
||||
}
|
||||
if a.Provisioning == nil {
|
||||
continue
|
||||
}
|
||||
for _, b := range a.Provisioning.Provisions {
|
||||
for _, b := range pl.Plan.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
|
||||
as = provider.ConsumerIdentity(pl.Node, pl.Module) // a plan that names none
|
||||
}
|
||||
want[as] = true
|
||||
}
|
||||
@@ -223,8 +196,8 @@ func (m *Moved) granted(planned map[string]bool) ([]contribution, map[string]boo
|
||||
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.
|
||||
// retire is one pass over what the plans bind and the grant file gives now: the logins only the grant file gives are
|
||||
// made and checked, as the grant path does. It retires nothing (ADR 0570 clause 5).
|
||||
func (m *Moved) retire(ctx context.Context) {
|
||||
h := m.h
|
||||
want, ok := m.asked(h.Node)
|
||||
@@ -245,7 +218,7 @@ func (m *Moved) retire(ctx context.Context) {
|
||||
}
|
||||
h.seed(ctx, want)
|
||||
h.apply(ctx, given)
|
||||
h.retireUnasked(ctx, want)
|
||||
// No login is retired for going unasked (novox/hq ADR 0570 clause 5): only Gone, on the operator's yes.
|
||||
h.r.lastWant = want
|
||||
for as := range h.failing {
|
||||
if !want[as] {
|
||||
@@ -261,13 +234,13 @@ func (m *Moved) retire(ctx context.Context) {
|
||||
|
||||
// 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) {
|
||||
err := bus.Watch(ctx, provisions.PlansBucket, func(key string, value []byte) {
|
||||
m.mu.Lock()
|
||||
defer m.mu.Unlock()
|
||||
if value == nil {
|
||||
delete(m.assignments, key)
|
||||
delete(m.plans, key)
|
||||
} else {
|
||||
m.assignments[key] = append([]byte(nil), value...)
|
||||
m.plans[key] = append([]byte(nil), value...)
|
||||
}
|
||||
})
|
||||
if err == nil {
|
||||
|
||||
@@ -11,7 +11,9 @@ import (
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
"git.novox.be/novox/mesh-sdk/go/catalogue"
|
||||
"git.novox.be/novox/mesh-sdk/go/provider"
|
||||
"git.novox.be/novox/mesh-sdk/go/provisions"
|
||||
)
|
||||
|
||||
const (
|
||||
@@ -57,6 +59,43 @@ func (b *bucketBus) Delete(_ context.Context, bucket, key string) error {
|
||||
return nil
|
||||
}
|
||||
|
||||
// putPlan is a consumer's plan on the seat mesh-provisioner, of these plan entries, each provision's key provided (novox/hq
|
||||
// ADR 0570 clauses 1, 2): what the helper binds by.
|
||||
func (b *bucketBus) putPlan(node, module string, entries ...map[string]any) {
|
||||
raw, _ := json.Marshal(entries)
|
||||
var ps []catalogue.PlanProvision
|
||||
if err := json.Unmarshal(raw, &ps); err != nil {
|
||||
panic(err)
|
||||
}
|
||||
for _, pp := range ps {
|
||||
pr := provisions.Provision{Provision: pp.Provision, Local: pp.Local, Consumer: provisions.End{Module: module, Node: node},
|
||||
Provider: provisions.End{Module: pp.ProviderModule, Node: pp.Provider}, State: provisions.Provided}
|
||||
key, _ := pr.Key()
|
||||
value, err := provisions.Encode(pr)
|
||||
if err != nil {
|
||||
panic(err)
|
||||
}
|
||||
_ = b.Put(context.Background(), provisions.ProvisionsBucket, key, value)
|
||||
}
|
||||
plan, err := provisions.EncodePlan(provisions.Plan{Node: node, Module: module, Plan: catalogue.Plan{Provisions: ps}})
|
||||
if err != nil {
|
||||
panic(err)
|
||||
}
|
||||
_ = b.Put(context.Background(), provisions.PlansBucket, node+"."+module, plan)
|
||||
}
|
||||
|
||||
// retireOn is a consumer's provision retiring with the operator's yes (ADR 0570 clause 5), its plan gone: the one way
|
||||
// a provider lets it go.
|
||||
func (b *bucketBus) retireOn(node, module, provides string) {
|
||||
pr := provisions.Provision{Provision: ProvisionName, Consumer: provisions.End{Module: module, Node: node},
|
||||
Provider: provisions.End{Module: provides, Node: "novox"}, State: provisions.Retiring, Why: module + " left",
|
||||
Warrant: &provisions.Warrant{Ask: "a-1", Option: "yes", By: "the operator"}}
|
||||
key, _ := pr.Key()
|
||||
value, _ := provisions.Encode(pr)
|
||||
_ = b.Delete(context.Background(), provisions.PlansBucket, node+"."+module)
|
||||
_ = b.Put(context.Background(), provisions.ProvisionsBucket, key, value)
|
||||
}
|
||||
|
||||
// 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) {
|
||||
@@ -65,12 +104,8 @@ func (b *bucketBus) bind(node, module string, contributes map[string]any) {
|
||||
|
||||
// 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)
|
||||
}
|
||||
b.putPlan(node, module, map[string]any{"provision": ProvisionName, "provider": "novox", "provider-module": provides,
|
||||
"contributes": contributes, "names": names})
|
||||
}
|
||||
|
||||
// consumersFile writes the file the machine's client fills: one credential per consumer.
|
||||
@@ -124,9 +159,9 @@ func newMovedWorldOf(t *testing.T, bound map[[2]string]string, spec MovedSpec) *
|
||||
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
|
||||
w.run.plans = map[string][]byte{}
|
||||
for k, v := range w.bus.buckets[provisions.PlansBucket] {
|
||||
w.run.plans[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,
|
||||
@@ -205,46 +240,27 @@ func TestTheConsumersFileMakesWhatTheGrantFilesMake(t *testing.T) {
|
||||
}
|
||||
}
|
||||
|
||||
// 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.
|
||||
// A consumer whose plan goes retires nothing by itself, however long (novox/hq ADR 0570 clause 5): its login is retired,
|
||||
// never dropped, and its entry removed only once its provision is retiring with the operator's yes.
|
||||
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)
|
||||
_ = w.bus.Delete(ctx, provisions.PlansBucket, "ace.app2")
|
||||
for i := 0; i < 8; i++ {
|
||||
w.pass()
|
||||
}
|
||||
w.now = w.now.Add(15 * time.Minute)
|
||||
w.now = w.now.Add(30 * 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")]))
|
||||
if len(w.a.removed) != 0 || w.h.r.waiting != nil {
|
||||
t.Fatalf("retired %v, waiting %v, with no warrant", w.a.removed, w.h.r.waiting)
|
||||
}
|
||||
|
||||
// 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 _, kept := w.bus.buckets[provider.BucketOf("postgres")]["ace.app2.postgres-database"]; !kept {
|
||||
t.Fatal("the entry went with no warrant")
|
||||
}
|
||||
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.bus.retireOn("ace", "app2", "postgres")
|
||||
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)
|
||||
@@ -254,35 +270,13 @@ func TestAConsumerMissingFromOneReadRetiresNothing(t *testing.T) {
|
||||
}
|
||||
}
|
||||
|
||||
// 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")
|
||||
w.bus.Delete(ctx, provisions.PlansBucket, "ace.app1")
|
||||
for i := 0; i < 8; i++ {
|
||||
w.pass()
|
||||
}
|
||||
@@ -310,12 +304,8 @@ func TestAConsumerWithoutACredentialIsWaitedFor(t *testing.T) {
|
||||
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)
|
||||
}
|
||||
w.bus.putPlan("ace", "claude-code", map[string]any{"provision": ProvisionName, "provider": "novox",
|
||||
"provider-module": "postgres", "as": agent})
|
||||
for _, o := range w.pass() {
|
||||
if o.Status != provider.Answered {
|
||||
t.Fatalf("%+v", o)
|
||||
@@ -353,10 +343,8 @@ func TestASpecWithoutGrantsMakesUnderThePlansAsAndIgnoresTheGrantFile(t *testing
|
||||
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)
|
||||
bus.putPlan("ace", "wiki", map[string]any{"provision": ProvisionName, "provider": "novox",
|
||||
"provider-module": "keycloak", "as": "wiki-client", "contributes": map[string]any{"label": "Wiki"}})
|
||||
if err := run.watch(ctx, bus); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
@@ -454,12 +442,8 @@ func TestTheConsumersFileMakesWhatTheGrantFileMakesForTheKeycloakShape(t *testin
|
||||
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)
|
||||
}
|
||||
w.bus.putPlan("home", "grafana-2", map[string]any{"provision": ProvisionName, "provider": "novox", "provider-module": "keycloak",
|
||||
"as": agent, "contributes": grafanaContributes, "names": grafanaNames})
|
||||
for _, o := range w.pass() {
|
||||
if o.Status != provider.Answered {
|
||||
t.Fatalf("%+v", o)
|
||||
|
||||
+152
-104
@@ -47,6 +47,7 @@ import (
|
||||
"time"
|
||||
|
||||
stdio "git.novox.be/novox/mesh-sdk/go"
|
||||
"git.novox.be/novox/mesh-sdk/go/provisions"
|
||||
)
|
||||
|
||||
// AssignmentsBucket is the controller's assignments mirror, `<node>.<module>`, whose value carries the plan of a
|
||||
@@ -124,6 +125,11 @@ type Entry struct {
|
||||
VaultName string `json:"vault-name,omitempty"`
|
||||
// Refused, when set, is why the consumer is not served, naming both ends.
|
||||
Refused string `json:"refused,omitempty"`
|
||||
// State is the provider's outcome for the consumer, which the holder of mesh-provisioner reads to set the
|
||||
// provision's key (novox/hq ADR 0570 clause 3): provided, failed with Why, or waits with what is awaited. Empty from
|
||||
// a provider before it.
|
||||
State string `json:"state,omitempty"`
|
||||
Why string `json:"why,omitempty"`
|
||||
// Contributes is what the consumer's module tells its provider about this provision (a database's name, its
|
||||
// extensions), from its plan's entry (issue 776). Handed to Make and Gone; never written to the bucket.
|
||||
Contributes map[string]any `json:"-"`
|
||||
@@ -185,11 +191,13 @@ type Options struct {
|
||||
type Provider struct {
|
||||
opts Options
|
||||
|
||||
mu sync.Mutex
|
||||
assignments map[string][]byte
|
||||
entries map[string][]byte
|
||||
said map[string]string
|
||||
poke chan struct{}
|
||||
mu sync.Mutex
|
||||
entries map[string][]byte
|
||||
// plans and provisions are the seat mesh-provisioner's states (ADR 0570 clauses 1, 2), by key.
|
||||
plans map[string][]byte
|
||||
provisions map[string][]byte
|
||||
said map[string]string
|
||||
poke chan struct{}
|
||||
}
|
||||
|
||||
// New is a provider answering its consumers as opts say.
|
||||
@@ -203,8 +211,8 @@ func New(opts Options) *Provider {
|
||||
if opts.Logf == nil {
|
||||
opts.Logf = func(f string, a ...any) { fmt.Printf("[provider] "+f+"\n", a...) }
|
||||
}
|
||||
return &Provider{opts: opts, assignments: map[string][]byte{}, entries: map[string][]byte{},
|
||||
said: map[string]string{}, poke: make(chan struct{}, 1)}
|
||||
return &Provider{opts: opts, entries: map[string][]byte{},
|
||||
plans: map[string][]byte{}, provisions: map[string][]byte{}, said: map[string]string{}, poke: make(chan struct{}, 1)}
|
||||
}
|
||||
|
||||
// What a pass did for one bind.
|
||||
@@ -232,16 +240,22 @@ func (p *Provider) Run(ctx context.Context) error {
|
||||
if err := p.watch(ctx, BucketOf(p.opts.Module), p.entries); err != nil {
|
||||
return fmt.Errorf("watching %s: %w", BucketOf(p.opts.Module), err)
|
||||
}
|
||||
if err := p.watch(ctx, AssignmentsBucket, p.assignments); err != nil {
|
||||
return fmt.Errorf("watching %s: %w", AssignmentsBucket, err)
|
||||
for _, bucket := range []string{provisions.PlansBucket, provisions.ProvisionsBucket} {
|
||||
into := p.plans
|
||||
if bucket == provisions.ProvisionsBucket {
|
||||
into = p.provisions
|
||||
}
|
||||
if err := p.watch(ctx, bucket, into); err != nil {
|
||||
return fmt.Errorf("watching %s: %w", bucket, err)
|
||||
}
|
||||
}
|
||||
tick := time.NewTicker(p.opts.Every)
|
||||
defer tick.Stop()
|
||||
for {
|
||||
p.mu.Lock()
|
||||
assignments, entries := clone(p.assignments), clone(p.entries)
|
||||
in := Inputs{Entries: clone(p.entries), Plans: clone(p.plans), Provisions: clone(p.provisions)}
|
||||
p.mu.Unlock()
|
||||
if _, err := p.pass(ctx, assignments, entries); err != nil {
|
||||
if _, err := p.passIn(ctx, in); err != nil {
|
||||
p.say("pass", err.Error())
|
||||
}
|
||||
select {
|
||||
@@ -262,7 +276,7 @@ func (p *Provider) watch(ctx context.Context, bucket string, into map[string][]b
|
||||
into[key] = append([]byte(nil), value...)
|
||||
}
|
||||
p.mu.Unlock()
|
||||
if bucket == AssignmentsBucket {
|
||||
if bucket != BucketOf(p.opts.Module) {
|
||||
select {
|
||||
case p.poke <- struct{}{}:
|
||||
default:
|
||||
@@ -302,110 +316,82 @@ func (p *Provider) Pass(ctx context.Context) ([]Outcome, error) {
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("reading %s: %w", BucketOf(p.opts.Module), err)
|
||||
}
|
||||
assignments, err := read(AssignmentsBucket)
|
||||
plans, err := read(provisions.PlansBucket)
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("reading %s: %w", AssignmentsBucket, err)
|
||||
return nil, fmt.Errorf("reading %s: %w", provisions.PlansBucket, err)
|
||||
}
|
||||
return p.pass(ctx, assignments, entries)
|
||||
provs, err := read(provisions.ProvisionsBucket)
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("reading %s: %w", provisions.ProvisionsBucket, err)
|
||||
}
|
||||
return p.passIn(ctx, Inputs{Entries: entries, Plans: plans, Provisions: provs})
|
||||
}
|
||||
|
||||
// planned is as much of an assignment as the helper reads: lenient, since the plan is the client's to act on.
|
||||
type planned struct {
|
||||
Node string `json:"node"`
|
||||
Module string `json:"module"`
|
||||
Provisioning *struct {
|
||||
Provisions []struct {
|
||||
Provision string `json:"provision"`
|
||||
Local string `json:"local"`
|
||||
Provider string `json:"provider"`
|
||||
ProviderModule string `json:"provider-module"`
|
||||
Contributes map[string]any `json:"contributes"`
|
||||
At string `json:"at"`
|
||||
From string `json:"from"`
|
||||
As string `json:"as"`
|
||||
Names map[string]string `json:"names"`
|
||||
CredentialForm string `json:"credential-form"`
|
||||
Name string `json:"name"`
|
||||
} `json:"provisions"`
|
||||
} `json:"provisioning"`
|
||||
// Inputs are what one pass reads: the provider's own entries, and the seat mesh-provisioner's plans and provisions
|
||||
// (novox/hq ADR 0570 clauses 1, 2), each by key. The controller's assignments are read no more (clause 2).
|
||||
type Inputs struct {
|
||||
Entries, Plans, Provisions map[string][]byte
|
||||
}
|
||||
|
||||
func (p *Provider) pass(ctx context.Context, assignments, entries map[string][]byte) ([]Outcome, error) {
|
||||
func (p *Provider) passIn(ctx context.Context, in Inputs) ([]Outcome, error) {
|
||||
entries := in.Entries
|
||||
var binds []Entry
|
||||
var out []Outcome
|
||||
unknown := map[string]bool{} // consumers whose assignment was skipped: their entries are not judged
|
||||
for _, key := range sortedKeys(assignments) {
|
||||
if !Standing(assignments[key]) {
|
||||
continue // ended or pending: read as a removed key (issue 1098)
|
||||
}
|
||||
var a planned
|
||||
if err := json.Unmarshal(assignments[key], &a); err != nil || a.Node+"."+a.Module != key {
|
||||
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)}
|
||||
// Every consumer is bound by its plan and its provision keys (ADR 0570 clauses 2, 3), and an entry no plan names is
|
||||
// never let go by the provider itself (clause 5): only a retiring key with the operator's yes retires it.
|
||||
seat := map[string]bool{}
|
||||
gated := map[string]provisions.Provision{}
|
||||
for _, key := range sortedKeys(in.Plans) {
|
||||
plan, err := provisions.DecodePlan(in.Plans[key])
|
||||
if err != nil || plan.Node+"."+plan.Module != key {
|
||||
o := Outcome{Key: key, Status: Failed, Why: fmt.Sprintf("the plan %s is not readable, so who it binds is not "+
|
||||
"known; it is skipped, and its entries are left as they are", key)}
|
||||
out = append(out, o)
|
||||
p.say(key, o.Status+" "+o.Why)
|
||||
unknown[key] = true
|
||||
unknown[key], seat[key] = true, true
|
||||
continue
|
||||
}
|
||||
if a.Provisioning == nil {
|
||||
continue
|
||||
}
|
||||
for _, b := range a.Provisioning.Provisions {
|
||||
seat[key] = true
|
||||
for _, b := range plan.Plan.Provisions {
|
||||
if b.Provider != p.opts.Node || b.ProviderModule != p.opts.Module {
|
||||
continue
|
||||
}
|
||||
binds = append(binds, Entry{Provider: p.opts.Node, ProviderModule: p.opts.Module,
|
||||
Provision: b.Provision, Consumer: a.Node, Module: a.Module, Local: b.Local,
|
||||
Contributes: b.Contributes, At: b.At, From: b.From, As: b.As, Names: b.Names,
|
||||
CredentialForm: b.CredentialForm, Name: b.Name})
|
||||
e := Entry{Provider: p.opts.Node, ProviderModule: p.opts.Module, Provision: b.Provision, Consumer: plan.Node,
|
||||
Module: plan.Module, Local: b.Local, Contributes: b.Contributes, At: b.At, From: b.From, As: b.As,
|
||||
Names: b.Names, CredentialForm: b.CredentialForm, Name: b.Name}
|
||||
binds = append(binds, e)
|
||||
}
|
||||
}
|
||||
for _, key := range sortedKeys(in.Provisions) {
|
||||
if pr, err := provisions.Decode(in.Provisions[key]); err == nil && pr.Provider.Node == p.opts.Node &&
|
||||
pr.Provider.Module == p.opts.Module {
|
||||
gated[key] = pr
|
||||
}
|
||||
}
|
||||
|
||||
want := map[string]bool{}
|
||||
for _, b := range binds {
|
||||
key := b.Key()
|
||||
want[key] = true
|
||||
o := p.answer(ctx, b, entries[key])
|
||||
out = append(out, o)
|
||||
p.say(key, o.Status+" "+o.Why)
|
||||
}
|
||||
for _, key := range sortedKeys(entries) {
|
||||
if want[key] || unknown[consumerOf(key)] {
|
||||
continue
|
||||
}
|
||||
var e Entry
|
||||
if err := json.Unmarshal(entries[key], &e); err != nil || e.Provider != p.opts.Node ||
|
||||
e.ProviderModule != p.opts.Module {
|
||||
continue // another machine's instance of this provider wrote it, or nothing of this helper's
|
||||
}
|
||||
if p.opts.Gone != nil {
|
||||
if err := p.opts.Gone(ctx, e); err != nil {
|
||||
o := Outcome{Key: key, Status: Failed, Why: "no longer bound, and the provider could not let it go: " +
|
||||
err.Error()}
|
||||
out = append(out, o)
|
||||
p.say(key, o.Status+" "+o.Why)
|
||||
continue
|
||||
}
|
||||
}
|
||||
if err := p.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()}
|
||||
if why, ok := p.ready(key, gated, in.Provisions); !ok {
|
||||
o := Outcome{Key: key, Status: Waiting, Why: why}
|
||||
out = append(out, o)
|
||||
p.say(key, o.Status+" "+o.Why)
|
||||
continue
|
||||
}
|
||||
out = append(out, Outcome{Key: key, Status: Removed})
|
||||
p.say(key, Removed)
|
||||
p.mu.Lock()
|
||||
delete(p.said, key)
|
||||
p.mu.Unlock()
|
||||
o := p.answer(ctx, b, entries[key])
|
||||
out = append(out, o)
|
||||
p.say(key, o.Status+" "+o.Why)
|
||||
}
|
||||
// ADR 0570 clause 5: a retiring key the operator said yes to is let go; one asked and not answered, or declined,
|
||||
// keeps its entry.
|
||||
for _, key := range sortedKeys(gated) {
|
||||
if pr := gated[key]; pr.State == provisions.Retiring && pr.Warrant != nil && pr.Warrant.Option == "yes" {
|
||||
want[key] = false
|
||||
if _, held := entries[key]; held {
|
||||
out = append(out, p.letGo(ctx, key, entries[key])...)
|
||||
}
|
||||
}
|
||||
}
|
||||
return out, nil
|
||||
}
|
||||
@@ -443,6 +429,7 @@ func (p *Provider) answer(ctx context.Context, e Entry, have []byte) Outcome {
|
||||
strings.Join(derivedKeys(offer.Serves), ", "))
|
||||
}
|
||||
if e.Refused != "" {
|
||||
e.State, e.Why = provisions.Failed, e.Refused
|
||||
return p.write(ctx, e, have, Outcome{Key: key, Status: Refused, Why: e.Refused})
|
||||
}
|
||||
|
||||
@@ -455,6 +442,7 @@ func (p *Provider) answer(ctx context.Context, e Entry, have []byte) Outcome {
|
||||
e.Refused = fmt.Sprintf("%s on %s serving %s to %s on %s: %v", e.ProviderModule, e.Provider, e.Provision,
|
||||
e.Module, e.Consumer, err)
|
||||
e.As = ""
|
||||
e.State, e.Why = provisions.Failed, e.Refused
|
||||
return p.write(ctx, e, have, Outcome{Key: key, Status: Refused, Why: e.Refused})
|
||||
}
|
||||
if e.At != "" || e.From != "" {
|
||||
@@ -506,11 +494,15 @@ func (p *Provider) answer(ctx context.Context, e Entry, have []byte) Outcome {
|
||||
}
|
||||
if p.opts.Make != nil {
|
||||
if err := p.opts.Make(ctx, Consumer{Entry: e, Credential: credential}); err != nil {
|
||||
why := err.Error()
|
||||
why = redact(why, credential)
|
||||
return Outcome{Key: key, Status: Failed, Why: "the provider could not make it: " + why}
|
||||
why := "the provider could not make it: " + redact(err.Error(), credential)
|
||||
// Said in its entry, which the holder of mesh-provisioner reads (ADR 0570 clause 3), with nothing served.
|
||||
failed := e
|
||||
failed.Values, failed.VaultName, failed.User, failed.State, failed.Why = nil, "", "", provisions.Failed, why
|
||||
p.write(ctx, failed, have, Outcome{Key: key, Status: Failed, Why: why})
|
||||
return Outcome{Key: key, Status: Failed, Why: why}
|
||||
}
|
||||
}
|
||||
e.State = provisions.Provided
|
||||
return p.write(ctx, e, have, Outcome{Key: key, Status: Answered})
|
||||
}
|
||||
|
||||
@@ -557,20 +549,20 @@ 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 }
|
||||
func SharedName(provider, provision string) string {
|
||||
return provisions.VaultName(provisions.FormShared, provisions.Provision{Provision: provision, Provider: provisions.End{Node: provider}})
|
||||
}
|
||||
|
||||
// HolderName is the vault's value of one holder's own credential, issued or given for it (mesh-vault perholder.go).
|
||||
func HolderName(provider, provision, node, module string) string {
|
||||
return "provision/" + provider + "/" + provision + "/" + node + "/" + module
|
||||
return provisions.VaultName(provisions.FormIssued, provisions.Provision{Provision: provision,
|
||||
Provider: provisions.End{Node: provider}, Consumer: provisions.End{Node: node, Module: module}})
|
||||
}
|
||||
|
||||
// PairName is the vault's value of one consumer's own login (ADR 0048, ADR 0158 §11, issue 677's note 12).
|
||||
func PairName(provider, provision, node, module, local string) string {
|
||||
n := "pair/" + provider + "/" + provision + "/" + node + "/" + module
|
||||
if local != "" {
|
||||
n += "/" + local
|
||||
}
|
||||
return n
|
||||
return provisions.VaultName(provisions.FormPair, provisions.Provision{Provision: provision, Local: local,
|
||||
Provider: provisions.End{Node: provider}, Consumer: provisions.End{Node: node, Module: module}})
|
||||
}
|
||||
|
||||
// identityUnusable is every character a consumer's identity does not hold (mesh-controller internal/catalogue
|
||||
@@ -672,8 +664,9 @@ func sortedKeys[V any](m map[string]V) []string {
|
||||
}
|
||||
|
||||
// StateBus is the Bus of a provider that runs as a module bundle over the runtime's stdio: the provider's own bucket,
|
||||
// BucketOf(module), is the module's state `bound`, and AssignmentsBucket is the state it reads,
|
||||
// `mesh-controller.assignments` (both declared in the module's manifest). Any other bucket is refused.
|
||||
// BucketOf(module), is the module's state `bound`, and provisions.PlansBucket and provisions.ProvisionsBucket are the
|
||||
// states it reads, `mesh-provisioner.plans` and `mesh-provisioner.provisions` (each declared in the module's manifest).
|
||||
// Any other bucket is refused.
|
||||
//
|
||||
// A stdio state watch lasts as long as the process and cannot be ended (stdio.KeptState.Watch), so Watch honours the
|
||||
// Bus contract by silencing instead: once ctx has ended the handler is never called again, though the runtime keeps
|
||||
@@ -687,8 +680,10 @@ func (b stateBus) state(bucket string) (stdio.KeptState, error) {
|
||||
switch bucket {
|
||||
case BucketOf(b.module):
|
||||
return stdio.State("bound"), nil
|
||||
case AssignmentsBucket:
|
||||
return stdio.State("mesh-controller.assignments"), nil
|
||||
case provisions.PlansBucket:
|
||||
return stdio.State(provisions.PlansRead), nil
|
||||
case provisions.ProvisionsBucket:
|
||||
return stdio.State(provisions.ProvisionsRead), nil
|
||||
}
|
||||
return stdio.KeptState{}, fmt.Errorf("%s is not a bucket this provider keeps or reads", bucket)
|
||||
}
|
||||
@@ -767,3 +762,56 @@ func fieldSet(f json.RawMessage) bool {
|
||||
s := strings.TrimSpace(string(f))
|
||||
return s != "" && s != "null" && s != `""`
|
||||
}
|
||||
|
||||
// letGo lets a consumer's entry go: the provider hears it gone (Gone, which retires what it kept for the consumer), and
|
||||
// the entry is removed while it is still as read.
|
||||
func (p *Provider) letGo(ctx context.Context, key string, held []byte) []Outcome {
|
||||
var e Entry
|
||||
if err := json.Unmarshal(held, &e); err != nil || e.Provider != p.opts.Node || e.ProviderModule != p.opts.Module {
|
||||
return nil // another machine's instance of this provider wrote it, or nothing of this helper's
|
||||
}
|
||||
said := func(o Outcome) []Outcome {
|
||||
p.say(key, o.Status+" "+o.Why)
|
||||
return []Outcome{o}
|
||||
}
|
||||
if p.opts.Gone != nil {
|
||||
if err := p.opts.Gone(ctx, e); err != nil {
|
||||
return said(Outcome{Key: key, Status: Failed, Why: "no longer bound, and the provider could not let it go: " + err.Error()})
|
||||
}
|
||||
}
|
||||
if err := p.remove(ctx, key, held); err != nil {
|
||||
if errors.Is(err, stdio.ErrChanged) {
|
||||
return said(Outcome{Key: key, Status: Superseded, Why: "another instance of the provider wrote its entry " +
|
||||
"since this pass read it, so it stands"})
|
||||
}
|
||||
return said(Outcome{Key: key, Status: Failed, Why: "no longer bound, and its entry could not be removed: " + err.Error()})
|
||||
}
|
||||
p.say(key, Removed)
|
||||
p.mu.Lock()
|
||||
delete(p.said, key)
|
||||
p.mu.Unlock()
|
||||
return []Outcome{{Key: key, Status: Removed}}
|
||||
}
|
||||
|
||||
// ready says whether the provision under key may be made now (novox/hq ADR 0570 clause 3): its key is on the seat, it is
|
||||
// asked or provided, and every provision it depends on is provided. Otherwise why it waits, naming the key or the
|
||||
// dependency.
|
||||
func (p *Provider) ready(key string, mine map[string]provisions.Provision, all map[string][]byte) (string, bool) {
|
||||
pr, ok := mine[key]
|
||||
if !ok {
|
||||
return "mesh-provisioner holds no key " + key + " naming this provider yet", false
|
||||
}
|
||||
if pr.State != provisions.Asked && pr.State != provisions.Provided {
|
||||
return "its key is " + pr.State + ": " + pr.Why, false
|
||||
}
|
||||
for _, d := range pr.DependsOn {
|
||||
dep, err := provisions.Decode(all[d])
|
||||
if err != nil {
|
||||
return "it depends on " + d + ", which mesh-provisioner holds no readable key of", false
|
||||
}
|
||||
if dep.State != provisions.Provided {
|
||||
return "it depends on " + d + ", which is " + dep.State, false
|
||||
}
|
||||
}
|
||||
return "", true
|
||||
}
|
||||
|
||||
+173
-46
@@ -15,6 +15,8 @@ import (
|
||||
"time"
|
||||
|
||||
stdio "git.novox.be/novox/mesh-sdk/go"
|
||||
"git.novox.be/novox/mesh-sdk/go/catalogue"
|
||||
"git.novox.be/novox/mesh-sdk/go/provisions"
|
||||
)
|
||||
|
||||
// standIn is a stand-in bus: buckets of keys and values, and watches that hand the current values first and then
|
||||
@@ -165,22 +167,63 @@ func (m *maker) Gone(_ context.Context, e Entry) error {
|
||||
return nil
|
||||
}
|
||||
|
||||
// assign puts a consumer's assignment with its plan's provisions into the stand-in mirror.
|
||||
func assign(t *testing.T, bus *standIn, node, module string, provisions ...map[string]string) {
|
||||
// assign puts a consumer's plan with its provisions on the seat mesh-provisioner, each provision's key provided, as its
|
||||
// holder writes them (novox/hq ADR 0570 clauses 1, 2): what a provider binds by.
|
||||
func assign(t *testing.T, bus *standIn, node, module string, binds ...map[string]string) {
|
||||
t.Helper()
|
||||
ps := []map[string]string{}
|
||||
ps = append(ps, provisions...)
|
||||
raw, err := json.Marshal(map[string]any{"node": node, "module": module, "assigned": "2026-10-15T00:00:00Z",
|
||||
"provisioning": map[string]any{"templates": map[string]string{"source": "x", "digest": "y"},
|
||||
"provisions": ps}})
|
||||
var ps []catalogue.PlanProvision
|
||||
for _, b := range binds {
|
||||
raw, _ := json.Marshal(b)
|
||||
var pp catalogue.PlanProvision
|
||||
if err := json.Unmarshal(raw, &pp); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
ps = append(ps, pp)
|
||||
}
|
||||
planned(t, bus, node, module, ps...)
|
||||
}
|
||||
|
||||
// planned puts a consumer's plan of these provisions on the seat, each provision's key provided.
|
||||
func planned(t *testing.T, bus *standIn, node, module string, ps ...catalogue.PlanProvision) {
|
||||
t.Helper()
|
||||
for _, pp := range ps {
|
||||
pr := provisions.Provision{Provision: pp.Provision, Local: pp.Local, Consumer: provisions.End{Module: module, Node: node},
|
||||
Provider: provisions.End{Module: pp.ProviderModule, Node: pp.Provider}, State: provisions.Provided}
|
||||
key, err := pr.Key()
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
value, err := provisions.Encode(pr)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
_ = bus.Put(context.Background(), provisions.ProvisionsBucket, key, value)
|
||||
}
|
||||
raw, err := provisions.EncodePlan(provisions.Plan{Node: node, Module: module, Plan: catalogue.Plan{Provisions: ps}})
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if err := bus.Put(context.Background(), AssignmentsBucket, node+"."+module, raw); err != nil {
|
||||
if err := bus.Put(context.Background(), provisions.PlansBucket, node+"."+module, raw); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
}
|
||||
|
||||
// retire marks a consumer's provision retiring, with the operator's yes (novox/hq ADR 0570 clause 5): the one way a
|
||||
// provider lets a consumer go.
|
||||
func retire(t *testing.T, bus *standIn, node, module, provision, provider, providerModule string) {
|
||||
t.Helper()
|
||||
pr := provisions.Provision{Provision: provision, Consumer: provisions.End{Module: module, Node: node},
|
||||
Provider: provisions.End{Module: providerModule, Node: provider}, State: provisions.Retiring, Why: module + " left " + node,
|
||||
Warrant: &provisions.Warrant{Ask: "a-1", Option: "yes", By: "the operator"}}
|
||||
key, _ := pr.Key()
|
||||
raw, err := provisions.Encode(pr)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
_ = bus.Delete(context.Background(), provisions.PlansBucket, node+"."+module)
|
||||
_ = bus.Put(context.Background(), provisions.ProvisionsBucket, key, raw)
|
||||
}
|
||||
|
||||
func bind(provision, local, provider, providerModule string) map[string]string {
|
||||
b := map[string]string{"provision": provision, "provider": provider, "provider-module": providerModule}
|
||||
if local != "" {
|
||||
@@ -352,7 +395,7 @@ func TestAnUnbindRemovesTheEntryAndLeavesAnotherProvidersAlone(t *testing.T) {
|
||||
if _, err := p.Pass(context.Background()); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
_ = bus.Delete(context.Background(), AssignmentsBucket, "ace.letta")
|
||||
retire(t, bus, "ace", "letta", "postgres-database", "novox", "postgres")
|
||||
out, err := p.Pass(context.Background())
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
@@ -418,11 +461,11 @@ func TestOnlyBindsNamingThisProviderAreAnswered(t *testing.T) {
|
||||
}
|
||||
}
|
||||
|
||||
// An assignment that cannot be read is skipped and said, by its key and why; the rest of the pass goes on (issue 825,
|
||||
// A plan 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"}`} {
|
||||
"another key": `{"v":1,"node":"g14","module":"umami","plan":{"templates":{"source":"","digest":""},"files":null}}`} {
|
||||
bus, m := newStandIn(), &maker{}
|
||||
v := &vault{values: map[string]string{"pair/novox/postgres-database/ace/letta": "s",
|
||||
"pair/novox/postgres-database/g14/umami": "u"}}
|
||||
@@ -431,7 +474,7 @@ func TestAnUnreadableAssignmentIsSkippedAndSaidAndTheRestGoesOn(t *testing.T) {
|
||||
if _, err := p.Pass(context.Background()); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
_ = bus.Put(context.Background(), AssignmentsBucket, "ace.letta", []byte(bad))
|
||||
_ = bus.Put(context.Background(), provisions.PlansBucket, "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...)) }
|
||||
@@ -457,7 +500,8 @@ func TestAnUnreadableAssignmentIsSkippedAndSaidAndTheRestGoesOn(t *testing.T) {
|
||||
}
|
||||
}
|
||||
|
||||
// An adapter that fails is said with the credential taken out of its words, and the entry is not written.
|
||||
// An adapter that fails is said with the credential taken out of its words, and its entry says it failed, serving
|
||||
// nothing (novox/hq ADR 0570 clause 3: the provider's outcome is its entry, which the holder of mesh-provisioner reads).
|
||||
func TestAFailedMakeIsSaidWithoutTheCredential(t *testing.T) {
|
||||
bus, m := newStandIn(), &maker{fail: errors.New(`password "hunter2-secret" rejected`)}
|
||||
v := &vault{values: map[string]string{"pair/novox/postgres-database/ace/letta": "hunter2-secret"}}
|
||||
@@ -475,8 +519,11 @@ func TestAFailedMakeIsSaidWithoutTheCredential(t *testing.T) {
|
||||
if strings.Contains(strings.Join(logged, "\n"), "hunter2") || len(logged) == 0 {
|
||||
t.Fatalf("logged %q", logged)
|
||||
}
|
||||
if bus.keys(BucketOf("postgres")) != nil {
|
||||
t.Fatal("an entry was written for a consumer the adapter could not make")
|
||||
var e Entry
|
||||
raw := bus.get(BucketOf("postgres"), "ace.letta.postgres-database")
|
||||
if err := json.Unmarshal(raw, &e); err != nil || e.State != "failed" || e.Values != nil || e.VaultName != "" ||
|
||||
strings.Contains(string(raw), "hunter2") {
|
||||
t.Fatalf("the failed entry is %s", raw)
|
||||
}
|
||||
}
|
||||
|
||||
@@ -492,7 +539,7 @@ func TestRunAnswersBindsAndUnbindsAsTheyAreWatched(t *testing.T) {
|
||||
go func() { done <- p.Run(ctx) }()
|
||||
assign(t, bus, "ace", "letta", bind("postgres-database", "", "novox", "postgres"))
|
||||
waitFor(t, func() bool { return bus.get(BucketOf("postgres"), "ace.letta.postgres-database") != nil })
|
||||
_ = bus.Delete(context.Background(), AssignmentsBucket, "ace.letta")
|
||||
retire(t, bus, "ace", "letta", "postgres-database", "novox", "postgres")
|
||||
waitFor(t, func() bool { return bus.get(BucketOf("postgres"), "ace.letta.postgres-database") == nil })
|
||||
cancel()
|
||||
if err := <-done; err != nil && !errors.Is(err, context.Canceled) {
|
||||
@@ -532,12 +579,12 @@ func TestAConsumersContributionIsHandedToTheProviderPerConsumer(t *testing.T) {
|
||||
bus, m := newStandIn(), &maker{}
|
||||
v := &vault{values: map[string]string{
|
||||
"pair/novox/postgres-database/ace/nextcloud": "a", "pair/novox/postgres-database/g14/umami": "b"}}
|
||||
raw := `{"node":"ace","module":"nextcloud","provisioning":{"templates":{"source":"x","digest":"y"},` +
|
||||
`"provisions":[{"provision":"postgres-database","provider":"novox","provider-module":"postgres",` +
|
||||
`"contributes":{"name":"nextcloud","extensions":["pg_trgm"]}}]}}`
|
||||
if err := bus.Put(context.Background(), AssignmentsBucket, "ace.nextcloud", []byte(raw)); err != nil {
|
||||
var pp catalogue.PlanProvision
|
||||
if err := json.Unmarshal([]byte(`{"provision":"postgres-database","provider":"novox","provider-module":"postgres",`+
|
||||
`"contributes":{"name":"nextcloud","extensions":["pg_trgm"]}}`), &pp); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
planned(t, bus, "ace", "nextcloud", pp)
|
||||
assign(t, bus, "g14", "umami", bind("postgres-database", "", "novox", "postgres"))
|
||||
if _, err := postgres(bus, v, m).Pass(context.Background()); err != nil {
|
||||
t.Fatal(err)
|
||||
@@ -561,12 +608,12 @@ func TestAConsumersEntryCarriesAtFromAndTheAsOfItsPlan(t *testing.T) {
|
||||
bus, m := newStandIn(), &maker{}
|
||||
v := &vault{values: map[string]string{
|
||||
"pair/novox/postgres-database/ace/claude-code": "a", "pair/novox/postgres-database/g14/umami": "b"}}
|
||||
raw := `{"node":"ace","module":"claude-code","provisioning":{"templates":{"source":"x","digest":"y"},` +
|
||||
`"provisions":[{"provision":"postgres-database","provider":"novox","provider-module":"postgres",` +
|
||||
`"at":"10.7.0.1","from":"novox","as":"mesh_ace_agent"}]}}`
|
||||
if err := bus.Put(context.Background(), AssignmentsBucket, "ace.claude-code", []byte(raw)); err != nil {
|
||||
var pp catalogue.PlanProvision
|
||||
if err := json.Unmarshal([]byte(`{"provision":"postgres-database","provider":"novox","provider-module":"postgres",`+
|
||||
`"at":"10.7.0.1","from":"novox","as":"mesh_ace_agent"}`), &pp); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
planned(t, bus, "ace", "claude-code", pp)
|
||||
assign(t, bus, "g14", "umami", bind("postgres-database", "", "novox", "postgres"))
|
||||
if _, err := postgres(bus, v, m).Pass(context.Background()); err != nil {
|
||||
t.Fatal(err)
|
||||
@@ -596,12 +643,12 @@ func TestAConsumersEntryCarriesTheNamesOfItsPlan(t *testing.T) {
|
||||
bus, m := newStandIn(), &maker{}
|
||||
v := &vault{values: map[string]string{
|
||||
"pair/novox/postgres-database/ace/gitea": "a", "pair/novox/postgres-database/g14/umami": "b"}}
|
||||
raw := `{"node":"ace","module":"gitea","provisioning":{"templates":{"source":"x","digest":"y"},` +
|
||||
`"provisions":[{"provision":"postgres-database","provider":"novox","provider-module":"postgres",` +
|
||||
`"names":{"name":"git.example.org","internal-name":"git.mesh","name-web":"web.example.org"}}]}}`
|
||||
if err := bus.Put(context.Background(), AssignmentsBucket, "ace.gitea", []byte(raw)); err != nil {
|
||||
var pp catalogue.PlanProvision
|
||||
if err := json.Unmarshal([]byte(`{"provision":"postgres-database","provider":"novox","provider-module":"postgres",`+
|
||||
`"names":{"name":"git.example.org","internal-name":"git.mesh","name-web":"web.example.org"}}`), &pp); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
planned(t, bus, "ace", "gitea", pp)
|
||||
assign(t, bus, "g14", "umami", bind("postgres-database", "", "novox", "postgres"))
|
||||
if _, err := postgres(bus, v, m).Pass(context.Background()); err != nil {
|
||||
t.Fatal(err)
|
||||
@@ -719,6 +766,26 @@ func TestThePlansCredentialFormWinsOverTheOffers(t *testing.T) {
|
||||
if e := run(Shared, nil); e.VaultName != "provision/novox/postgres-database" {
|
||||
t.Fatalf("the offer's form is the default: %+v", e)
|
||||
}
|
||||
{
|
||||
// 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).
|
||||
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)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// An instance deletes only an entry it wrote, and only while the entry is as it read it: one that another instance of
|
||||
@@ -733,7 +800,7 @@ func TestAnEntryAnotherInstanceWroteAfterTheMoveStands(t *testing.T) {
|
||||
}
|
||||
// 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")
|
||||
retire(t, bus, "ace", "letta", "postgres-database", "novox", "postgres")
|
||||
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 {
|
||||
@@ -781,23 +848,83 @@ func TestAFailedMakeIsSaidWithoutTheCredentialInAnyOfItsForms(t *testing.T) {
|
||||
}
|
||||
}
|
||||
|
||||
// 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)
|
||||
// novox/hq ADR 0570 clauses 3 and 5 (issue 1954 L6-P3, P4): a consumer the seat mesh-provisioner plans is bound by its
|
||||
// plan and its provision key, never by its assignment. Its entry is made only once its key is asked and every provision
|
||||
// it depends on is provided, and says provided; a plan gone retires nothing by itself; a retiring key is let go only on
|
||||
// the operator's yes.
|
||||
func TestTheSeatsKeysSayWhenAProviderActsAndWhenItLetsGo(t *testing.T) {
|
||||
ctx := context.Background()
|
||||
bus, m := newStandIn(), &maker{}
|
||||
v := &vault{values: map[string]string{"pair/novox/postgres-database/ace/letta": "s1"}}
|
||||
// The same consumer's assignment names another provider: the seat's plan wins.
|
||||
assign(t, bus, "ace", "letta", bind("postgres-database", "", "shanks", "postgres"))
|
||||
plan, _ := provisions.EncodePlan(provisions.Plan{Node: "ace", Module: "letta", Plan: catalogue.Plan{
|
||||
Provisions: []catalogue.PlanProvision{{Provision: "postgres-database", Provider: "novox", ProviderModule: "postgres"}}}})
|
||||
_ = bus.Put(ctx, provisions.PlansBucket, "ace.letta", plan)
|
||||
key := "ace.letta.postgres-database"
|
||||
put := func(k string, state string, depends []string, warrant *provisions.Warrant) {
|
||||
pr := provisions.Provision{Provision: "postgres-database", Consumer: provisions.End{Module: "letta", Node: "ace"},
|
||||
Provider: provisions.End{Module: "postgres", Node: "novox"}, DependsOn: depends, State: state, Warrant: warrant}
|
||||
if k != key {
|
||||
_, _, pr.Provision, _, _ = provisions.ParseKey(k)
|
||||
pr.Provider = provisions.End{Module: "mesh-vault", Node: "novox"}
|
||||
}
|
||||
if state != provisions.Asked && state != provisions.Provided {
|
||||
pr.Why = "said"
|
||||
}
|
||||
raw, err := provisions.Encode(pr)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
_ = bus.Put(ctx, provisions.ProvisionsBucket, k, raw)
|
||||
}
|
||||
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)
|
||||
p := postgres(bus, v, m)
|
||||
steps := []struct {
|
||||
name string
|
||||
before func()
|
||||
status string
|
||||
says string
|
||||
entry bool
|
||||
}{
|
||||
{name: "no key yet", status: Waiting, says: "holds no key"},
|
||||
{name: "a dependency asked", before: func() {
|
||||
put("ace.letta.secret", provisions.Asked, nil, nil)
|
||||
put(key, provisions.Asked, []string{"ace.letta.secret"}, nil)
|
||||
}, status: Waiting, says: "depends on ace.letta.secret, which is asked"},
|
||||
{name: "the dependency provided", before: func() { put("ace.letta.secret", provisions.Provided, nil, nil) },
|
||||
status: Answered, entry: true},
|
||||
{name: "its plan gone: nothing let go by itself", before: func() { _ = bus.Delete(ctx, provisions.PlansBucket, "ace.letta") },
|
||||
entry: true},
|
||||
{name: "retiring, not yet answered", before: func() { put(key, provisions.Retiring, nil, nil) }, entry: true},
|
||||
{name: "retiring, the operator's yes", before: func() {
|
||||
put(key, provisions.Retiring, nil, &provisions.Warrant{Ask: "a-1", Option: "yes", By: "the operator"})
|
||||
}, status: Removed},
|
||||
}
|
||||
for _, s := range steps {
|
||||
if s.before != nil {
|
||||
s.before()
|
||||
}
|
||||
out, err := p.Pass(ctx)
|
||||
if err != nil {
|
||||
t.Fatalf("%s: %v", s.name, err)
|
||||
}
|
||||
var mine []Outcome
|
||||
for _, o := range out {
|
||||
if o.Key == key {
|
||||
mine = append(mine, o)
|
||||
}
|
||||
}
|
||||
if s.status != "" && (len(mine) != 1 || mine[0].Status != s.status || !strings.Contains(mine[0].Why, s.says)) {
|
||||
t.Errorf("%s: %+v", s.name, mine)
|
||||
}
|
||||
if held := bus.get(BucketOf("postgres"), key) != nil; held != s.entry {
|
||||
t.Errorf("%s: an entry held %v", s.name, held)
|
||||
}
|
||||
if s.entry && entryOf(t, bus, key).State != provisions.Provided {
|
||||
t.Errorf("%s: the entry does not say provided", s.name)
|
||||
}
|
||||
}
|
||||
if !slices.Equal(m.gone, []string{key}) || m.made[key] != "s1" {
|
||||
t.Errorf("gone %v, made %v: let go once, on the warrant, and made from the vault's value", m.gone, m.made)
|
||||
}
|
||||
}
|
||||
|
||||
@@ -10,6 +10,7 @@ import (
|
||||
"time"
|
||||
|
||||
stdio "git.novox.be/novox/mesh-sdk/go"
|
||||
"git.novox.be/novox/mesh-sdk/go/provisions"
|
||||
)
|
||||
|
||||
// runtime drives a bundle over pipes as the runtime does: it reads what the bundle says and sends what the runtime would.
|
||||
@@ -55,8 +56,9 @@ func serve(t *testing.T, tools []stdio.Tool) *runtime {
|
||||
return &runtime{t: t, lines: bufio.NewScanner(outR), in: inW}
|
||||
}
|
||||
|
||||
// The bus of a moved provider maps its own bucket to the state `bound` and the assignments to the state
|
||||
// `mesh-controller.assignments`, writes only the own bucket's state, and refuses any other bucket.
|
||||
// The bus of a moved provider maps its own bucket to the state `bound` and the seat's plans to the state
|
||||
// `mesh-provisioner.plans` (novox/hq ADR 0570 clause 2), writes only the own bucket's state, and refuses any other
|
||||
// bucket, the controller's assignments among them.
|
||||
func TestStateBusMapsTheProvidersBucketsToTheRuntimesStates(t *testing.T) {
|
||||
bus := StateBus("postgres")
|
||||
rt := serve(t, []stdio.Tool{
|
||||
@@ -72,7 +74,7 @@ func TestStateBusMapsTheProvidersBucketsToTheRuntimesStates(t *testing.T) {
|
||||
})
|
||||
for i, c := range []struct{ bucket, state string }{
|
||||
{BucketOf("postgres"), "bound"},
|
||||
{AssignmentsBucket, "mesh-controller.assignments"},
|
||||
{provisions.PlansBucket, "mesh-provisioner.plans"},
|
||||
} {
|
||||
rt.call(i+1, "watch", map[string]any{"bucket": c.bucket})
|
||||
w := rt.next()
|
||||
@@ -97,7 +99,7 @@ func TestStateBusMapsTheProvidersBucketsToTheRuntimesStates(t *testing.T) {
|
||||
rt.send(map[string]any{"jsonrpc": "2.0", "id": d["id"], "result": map[string]any{}})
|
||||
rt.next()
|
||||
|
||||
for _, bucket := range []string{"other_bound", "elsewhere"} {
|
||||
for _, bucket := range []string{"other_bound", "elsewhere", AssignmentsBucket} {
|
||||
if err := bus.Watch(context.Background(), bucket, func(string, []byte) {}); err == nil || !strings.Contains(err.Error(), "not a bucket") {
|
||||
t.Fatalf("a watch of %s gave %v", bucket, err)
|
||||
}
|
||||
@@ -118,7 +120,7 @@ func TestStateBusWatchCallsItsHandlerNoMoreOnceItsContextEnded(t *testing.T) {
|
||||
handled := make(chan string, 8)
|
||||
rt := serve(t, []stdio.Tool{
|
||||
{Name: "start", Run: func(map[string]any) (any, error) {
|
||||
return "started", bus.Watch(ctx, AssignmentsBucket, func(key string, value []byte) {
|
||||
return "started", bus.Watch(ctx, provisions.PlansBucket, func(key string, value []byte) {
|
||||
handled <- key + "=" + string(value)
|
||||
})
|
||||
}},
|
||||
|
||||
Reference in New Issue
Block a user