Drop artifact-store-tools' controller client; the provisioner client reads placement
The controller is not placed and its assignments mirror is frozen (ADR 0590). artifact-store-tools already answers through the seat's holder, so its controller client goes. mesh-provisioner-client reads the seat mesh-placement's placement, which its account is granted whole: a module placed here stays while the seat holds no plan of it, filled from its last plan, and leaves only when neither names it (issue 1126 B12). Claude-Session: https://claude.ai/code/session_01NcbgfJ9Gum47zkFKWRaovL
This commit is contained in:
@@ -153,7 +153,7 @@ func TestOlderNamesNotReadAreSaidFirst(t *testing.T) {
|
||||
func references(t *testing.T, a *asked) func(map[string]any) (any, error) {
|
||||
f := world()
|
||||
s := &Store{Container: "mesh-registry", Run: f.run(t)}
|
||||
for _, tl := range Tools(s, Controller{Ask: a.ask}, &Catalogue{Ask: a.ask}) {
|
||||
for _, tl := range Tools(s, recordsAsked{a}, &Catalogue{Ask: a.ask}) {
|
||||
if tl.Name == "artifact_store_references" {
|
||||
return tl.Run
|
||||
}
|
||||
|
||||
@@ -2,22 +2,16 @@ package main
|
||||
|
||||
import (
|
||||
"encoding/json"
|
||||
"errors"
|
||||
"fmt"
|
||||
"strconv"
|
||||
"strings"
|
||||
)
|
||||
|
||||
// What the records say, asked of the controller and never copied (novox/hq ADR 0251 §2). The
|
||||
// controller decides what the mesh keeps; this bundle only sets that beside what the store holds.
|
||||
// What the records say, in the shape the seat's holder answers them (Holder; novox/hq ADR 0251 §2, ADR 0571 clause 2):
|
||||
// the holder decides what the mesh keeps; the tools only set that beside what the store holds.
|
||||
|
||||
// Asking is how a tool on the mesh is asked: stdio.Ask, or a test's.
|
||||
type Asking func(key string, body any) (json.RawMessage, error)
|
||||
|
||||
// ControllerSeat is the seat whose verbs the controller serves.
|
||||
const ControllerSeat = "mesh-controller"
|
||||
|
||||
// Record is one reference the mesh recorded making, as the controller's `artifacts` answers it.
|
||||
// Record is one reference the mesh recorded making, as the seat's `artifacts` answers it.
|
||||
type Record struct {
|
||||
Reference string `json:"reference"`
|
||||
Repository string `json:"repository,omitempty"`
|
||||
@@ -27,7 +21,7 @@ type Record struct {
|
||||
Why []string `json:"why,omitempty"`
|
||||
}
|
||||
|
||||
// Records is the controller's `artifacts` answer.
|
||||
// Records is the seat's `artifacts` answer.
|
||||
type Records struct {
|
||||
Store string `json:"store"`
|
||||
KeptBuilds int `json:"kept_builds"`
|
||||
@@ -39,7 +33,7 @@ type Records struct {
|
||||
Note string `json:"-"`
|
||||
}
|
||||
|
||||
// CollectAnswer is the controller's `collect` answer.
|
||||
// CollectAnswer is the seat's `collect` answer.
|
||||
type CollectAnswer struct {
|
||||
DryRun bool `json:"dry_run"`
|
||||
Store string `json:"store"`
|
||||
@@ -53,15 +47,12 @@ type CollectAnswer struct {
|
||||
Skipped int `json:"skipped"`
|
||||
Left int `json:"left"`
|
||||
Stopped string `json:"stopped"`
|
||||
// Rest is whatever else the controller said, passed on as it said it.
|
||||
// Rest is whatever else the sweep said, passed on as it said it.
|
||||
Rest map[string]json.RawMessage `json:"-"`
|
||||
}
|
||||
|
||||
// Controller asks the controller's seat through the runtime.
|
||||
type Controller struct{ Ask Asking }
|
||||
|
||||
// answerOf reads a verb's answer whatever wraps it: the controller's {output, ok, answer}, a text the
|
||||
// runtime handed over, or the protocol's content list.
|
||||
// answerOf reads a verb's answer whatever wraps it: {output, ok, answer}, a text the runtime handed over, or the
|
||||
// protocol's content list.
|
||||
func answerOf(raw json.RawMessage) (json.RawMessage, string, bool) {
|
||||
var s string
|
||||
if json.Unmarshal(raw, &s) == nil {
|
||||
@@ -95,89 +86,11 @@ func answerOf(raw json.RawMessage) (json.RawMessage, string, bool) {
|
||||
return raw, string(raw), true
|
||||
}
|
||||
|
||||
func (c Controller) verb(name string, args map[string]any) (json.RawMessage, error) {
|
||||
if c.Ask == nil {
|
||||
return nil, errors.New("this bundle has no way to ask the controller")
|
||||
}
|
||||
raw, err := c.Ask("seat:"+ControllerSeat+"."+name, args)
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("the controller's %s could not be asked: %w", name, err)
|
||||
}
|
||||
answer, output, ok := answerOf(raw)
|
||||
if !ok {
|
||||
return nil, fmt.Errorf("the controller refused %s: %s", name, lastLine(output))
|
||||
}
|
||||
if len(answer) == 0 || string(answer) == "null" {
|
||||
return nil, fmt.Errorf("the controller's %s answered no data: %s", name, lastLine(output))
|
||||
}
|
||||
return answer, nil
|
||||
}
|
||||
|
||||
func lastLine(s string) string {
|
||||
lines := strings.Split(strings.TrimSpace(s), "\n")
|
||||
return strings.TrimSpace(lines[len(lines)-1])
|
||||
}
|
||||
|
||||
// Artifacts asks what the records say, with the references already let go of when the answer can carry
|
||||
// them; when it cannot, without, and says so.
|
||||
func (c Controller) Artifacts(repository string) (*Records, error) {
|
||||
args := map[string]any{"collected": "true"}
|
||||
if repository != "" {
|
||||
args["repository"] = repository
|
||||
}
|
||||
raw, err := c.verb("artifacts", args)
|
||||
collected := true
|
||||
note := ""
|
||||
if err != nil {
|
||||
delete(args, "collected")
|
||||
var again error
|
||||
raw, again = c.verb("artifacts", args)
|
||||
if again != nil {
|
||||
return nil, again
|
||||
}
|
||||
collected = false
|
||||
note = "the references the mesh already let go of could not be asked for (" + err.Error() +
|
||||
"), so a manifest the mesh let go of that the store still holds is counted as unrecorded"
|
||||
}
|
||||
var r Records
|
||||
if err := json.Unmarshal(raw, &r); err != nil {
|
||||
return nil, fmt.Errorf("the controller's artifacts answer is not readable: %v", err)
|
||||
}
|
||||
for i := range r.References {
|
||||
fill(&r.References[i])
|
||||
}
|
||||
r.CollectedAsked, r.Note = collected, note
|
||||
return &r, nil
|
||||
}
|
||||
|
||||
// Collect asks the controller's sweep: a dry run unless confirm, which needs why.
|
||||
func (c Controller) Collect(why string, confirm bool, most int) (*CollectAnswer, error) {
|
||||
args := map[string]any{}
|
||||
if confirm {
|
||||
if strings.TrimSpace(why) == "" {
|
||||
return nil, errors.New("a real collection needs why: nothing was asked")
|
||||
}
|
||||
args["why"] = why
|
||||
args["confirm"] = "true"
|
||||
}
|
||||
if most > 0 {
|
||||
args["most"] = strconv.Itoa(most)
|
||||
}
|
||||
raw, err := c.verb("collect", args)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
var a CollectAnswer
|
||||
if err := json.Unmarshal(raw, &a); err != nil {
|
||||
return nil, fmt.Errorf("the controller's collect answer is not readable: %v", err)
|
||||
}
|
||||
_ = json.Unmarshal(raw, &a.Rest)
|
||||
if !confirm && !a.DryRun && len(a.LetGo) > 0 {
|
||||
return nil, fmt.Errorf("the controller let go of %d artifacts when only a dry run was asked", len(a.LetGo))
|
||||
}
|
||||
return &a, nil
|
||||
}
|
||||
|
||||
// fill reads the repository, digest and kind out of a reference when the answer left them out.
|
||||
func fill(r *Record) {
|
||||
repo, digest, kind := splitReference(r.Reference)
|
||||
|
||||
@@ -9,6 +9,7 @@ import (
|
||||
"os"
|
||||
"reflect"
|
||||
"sort"
|
||||
"strconv"
|
||||
"strings"
|
||||
"sync"
|
||||
"testing"
|
||||
@@ -280,7 +281,7 @@ func TestOnlyTheirBytesLeaveOutWhatAnotherStateMarks(t *testing.T) {
|
||||
}
|
||||
}
|
||||
|
||||
// asked records every ask, and answers as the controller would.
|
||||
// asked records every ask, and answers as the seat and the catalogue would.
|
||||
type asked struct {
|
||||
mu sync.Mutex
|
||||
calls []string
|
||||
@@ -293,6 +294,51 @@ type asked struct {
|
||||
noCatalogue bool
|
||||
}
|
||||
|
||||
// recordsAsked is the tools' Recorded in a test: each call asked of asked, as `artifacts` or `collect`, with the
|
||||
// arguments a real run gives the seat's sweep.
|
||||
type recordsAsked struct{ a *asked }
|
||||
|
||||
func (r recordsAsked) Artifacts(repository string) (*Records, error) {
|
||||
args := map[string]any{"collected": "true"}
|
||||
if repository != "" {
|
||||
args["repository"] = repository
|
||||
}
|
||||
raw, err := r.a.ask("artifacts", args)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
var recs Records
|
||||
answer, _, _ := answerOf(raw)
|
||||
if err := json.Unmarshal(answer, &recs); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
for i := range recs.References {
|
||||
fill(&recs.References[i])
|
||||
}
|
||||
recs.CollectedAsked = true
|
||||
return &recs, nil
|
||||
}
|
||||
|
||||
func (r recordsAsked) Collect(why string, confirm bool, most int) (*CollectAnswer, error) {
|
||||
args := map[string]any{}
|
||||
if confirm {
|
||||
if strings.TrimSpace(why) == "" {
|
||||
return nil, fmt.Errorf("a real collection needs why: nothing was asked")
|
||||
}
|
||||
args["why"], args["confirm"] = why, "true"
|
||||
}
|
||||
if most > 0 {
|
||||
args["most"] = strconv.Itoa(most)
|
||||
}
|
||||
raw, err := r.a.ask("collect", args)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
var c CollectAnswer
|
||||
answer, _, _ := answerOf(raw)
|
||||
return &c, json.Unmarshal(answer, &c)
|
||||
}
|
||||
|
||||
func (a *asked) ask(key string, body any) (json.RawMessage, error) {
|
||||
a.mu.Lock()
|
||||
defer a.mu.Unlock()
|
||||
@@ -304,7 +350,7 @@ func (a *asked) ask(key string, body any) (json.RawMessage, error) {
|
||||
}
|
||||
var answer any
|
||||
switch key {
|
||||
case "seat:mesh-controller.artifacts":
|
||||
case "artifacts":
|
||||
answer = records()
|
||||
case CatalogueModules:
|
||||
if a.noCatalogue {
|
||||
@@ -331,7 +377,7 @@ func (a *asked) ask(key string, body any) (json.RawMessage, error) {
|
||||
// The catalogue's tool answers through the runtime as the protocol's content list.
|
||||
raw, _ := json.Marshal(map[string]any{"module": map[string]any{"module": name, "manifest": json.RawMessage(manifest)}})
|
||||
return json.Marshal(map[string]any{"content": []map[string]any{{"type": "text", "text": string(raw)}}})
|
||||
case "seat:mesh-controller.collect":
|
||||
case "collect":
|
||||
if args["confirm"] == "true" {
|
||||
answer = CollectAnswer{LetGo: []string{"artifact-store://app/server@" + d("m-old")}}
|
||||
} else {
|
||||
@@ -347,7 +393,7 @@ func (a *asked) ask(key string, body any) (json.RawMessage, error) {
|
||||
func tool(t *testing.T, a *asked, name string) func(map[string]any) (any, error) {
|
||||
f := world()
|
||||
s := &Store{Container: "mesh-registry", Run: f.run(t)}
|
||||
for _, tl := range Tools(s, Controller{Ask: a.ask}, &Catalogue{Ask: a.ask}) {
|
||||
for _, tl := range Tools(s, recordsAsked{a}, &Catalogue{Ask: a.ask}) {
|
||||
if tl.Name == name {
|
||||
return tl.Run
|
||||
}
|
||||
@@ -390,21 +436,6 @@ func TestARealRunWithoutWhyIsRefusedAndAsksNothing(t *testing.T) {
|
||||
}
|
||||
}
|
||||
|
||||
func TestReferencesFallBackWhenCollectedCannotBeCarried(t *testing.T) {
|
||||
a := &asked{fail: map[string]bool{"collected": true}}
|
||||
out, err := tool(t, a, "artifact_store_references")(map[string]any{"repository": "app/server"})
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
m := out.(map[string]any)
|
||||
if m["note"] == nil || m["manifests"] == nil {
|
||||
t.Errorf("answer %v", m)
|
||||
}
|
||||
if _, err := tool(t, a, "artifact_store_references")(map[string]any{"repository": "nope"}); err == nil {
|
||||
t.Error("an unknown repository was answered")
|
||||
}
|
||||
}
|
||||
|
||||
func TestTheToolsServedAreTheToolsTheManifestNames(t *testing.T) {
|
||||
raw, err := os.ReadFile("../../module.json")
|
||||
if err != nil {
|
||||
@@ -423,7 +454,7 @@ func TestTheToolsServedAreTheToolsTheManifestNames(t *testing.T) {
|
||||
t.Fatal(err)
|
||||
}
|
||||
var served []string
|
||||
for _, tl := range Tools(&Store{}, Controller{}, &Catalogue{}) {
|
||||
for _, tl := range Tools(&Store{}, recordsAsked{&asked{}}, &Catalogue{}) {
|
||||
if !strings.HasPrefix(tl.Name, "artifact_store_") || tl.Description == "" || tl.Run == nil {
|
||||
t.Errorf("tool %q", tl.Name)
|
||||
}
|
||||
|
||||
@@ -12,7 +12,7 @@ import (
|
||||
)
|
||||
|
||||
// Recorded is what says which references the mesh keeps and lets go of them: the seat's holder (Holder, novox/hq ADR
|
||||
// 0571 clause 2), or before it the controller (Controller), each in the controller's answer's shape.
|
||||
// 0571 clause 2), or a test's stand-in.
|
||||
type Recorded interface {
|
||||
Artifacts(repository string) (*Records, error)
|
||||
Collect(why string, confirm bool, most int) (*CollectAnswer, error)
|
||||
|
||||
@@ -3,16 +3,15 @@
|
||||
Holds the `node-provisioner` seat on every node: the provisioner's client (novox/hq ADR 0349 clauses 3 to 5, design 57
|
||||
P5). Go, one binary (`cmd/mesh-provisioner-client`), run as a service of its own under the account
|
||||
`mesh-provisioner-client`, outside the tool runner, on a bus account of its own, `provisioner.<node>`, delivered as
|
||||
its own secret `provisioner-broker` (mesh-controller `internal/broker/provisioner.go`). The bus grants that account
|
||||
its own machine's keys and, to read alone, the seat mesh-provisioner's whole `plans` and the controller's whole
|
||||
assignments mirror, where a provider's consumers on any machine are bound to it (ADR 0373); nothing else.
|
||||
its own secret `provisioner-broker` (mesh-sdk `buscompose/provisioner.go`). The bus grants that account its own
|
||||
machine's keys and, to read alone, the seat mesh-provisioner's whole `plans`, where a provider's consumers on any
|
||||
machine are bound to it (ADR 0373), and the seat mesh-placement's whole `placement`; nothing else.
|
||||
|
||||
## What it does
|
||||
|
||||
For each module placed on its node — a plan in the seat mesh-provisioner's `plans` (`mesh-provisioner_plans`,
|
||||
`<node>.<module>`, novox/hq ADR 0570 clause 2), or, for a module the seat holds no plan of, until the controller hands
|
||||
its lease away, an assignment in `mesh-controller_assignments` whose value carries a `provisioning` plan; a module
|
||||
leaves only when neither places it: the templates archive
|
||||
`<node>.<module>`, novox/hq ADR 0570 clause 2); a module the seat mesh-placement still places here stays while the
|
||||
seat holds no plan of it, filled from its last plan, and leaves only when neither names it (issue 1126): the templates archive
|
||||
by reference and digest, the files it fills (path, template, `owner-at-runtime`, how its program takes a change)
|
||||
and the placement values its templates name — the client:
|
||||
|
||||
|
||||
@@ -34,24 +34,25 @@ import (
|
||||
// back. A module taken off the node loses its needs and its report; the node-engine removes the files the
|
||||
// declaration no longer names.
|
||||
//
|
||||
// The owners it fills from are placement (the controller's, in the module's assignment), config
|
||||
// (mesh-config_resolved) and the vault, asked for each secret the module's plan names (secrets.go), and for each
|
||||
// consumer's credential a provider here lists (consumers.go); and each provider's entries for a consumer here. Every
|
||||
// key it writes is its own machine's, and every key it reads but the assignments mirror's, which it reads whole for
|
||||
// who binds a provider here (ADR 0373): its bus account is granted no other.
|
||||
// The owners it fills from are placement (in the module's plan), config (mesh-config_resolved) and the vault, asked for
|
||||
// each secret the module's plan names (secrets.go), and for each consumer's credential a provider here lists
|
||||
// (consumers.go); and each provider's entries for a consumer here. Every key it writes is its own machine's; it reads
|
||||
// the plans whole, for who binds a provider here (ADR 0373), and the seat mesh-placement's placement, for which
|
||||
// modules stay here: its bus account is granted no other.
|
||||
|
||||
// The buckets the client reads and writes, and the subject it reports on.
|
||||
const (
|
||||
// PlansBucket is where each module's provisioning plan on each node is read: the seat mesh-provisioner's
|
||||
// `plans` (novox/hq ADR 0570 clause 2), read as an assignment's value carried it (asAssignment).
|
||||
PlansBucket = provisions.PlansBucket
|
||||
// ControllerAssignmentsBucket is the controller's assignments mirror, read for a module the seat holds no plan of
|
||||
// until the controller hands its lease away: the holder writes plans only for modules with a want, so a module
|
||||
// that runs without one (nats on novox) would otherwise lose its files' fill (review of 2026-10-30).
|
||||
ControllerAssignmentsBucket = "mesh-controller_assignments"
|
||||
ConfigBucket = "mesh-config_resolved"
|
||||
RegistryBucket = "mesh-provisioner_registry"
|
||||
AppliedBucket = "mesh-provisioner_applied"
|
||||
// PlacementBucket is the seat mesh-placement's placement (novox/hq ADR 0567): a module it names on this node stays
|
||||
// here while the seat mesh-provisioner holds no plan of it (nats on novox runs without a want), filled from its
|
||||
// last plan, and leaves only when neither names it (issue 1126 B12; the controller's assignments mirror, read for
|
||||
// this before, is frozen since the controller's end, ADR 0590).
|
||||
PlacementBucket = delivery.PlacementBucket
|
||||
ConfigBucket = "mesh-config_resolved"
|
||||
RegistryBucket = "mesh-provisioner_registry"
|
||||
AppliedBucket = "mesh-provisioner_applied"
|
||||
)
|
||||
|
||||
// AppliedSubject is where the client says what it applied for one module (ADR 0349 clauses 4 and 15).
|
||||
@@ -152,15 +153,17 @@ type Client struct {
|
||||
type state struct {
|
||||
registry map[string]Answer
|
||||
plans map[string]Plan
|
||||
// seatPlans and controllerAssigned are each source's value by key, as an assignment's value reads: the seat's
|
||||
// plan wins, and the controller's assignment stands for a module the seat has no plan of.
|
||||
seatPlans, controllerAssigned map[string][]byte
|
||||
refused map[string]string // a module whose plan cannot be acted on, and why
|
||||
config map[string]map[string]json.RawMessage
|
||||
applied map[string]bool // module → a report is kept
|
||||
reported map[string][]byte // module → the last report said, undated
|
||||
last map[string]Report // module → the last report said, dated
|
||||
secrets map[string]heldSecret // the vault's name → what this run holds of it; never written anywhere
|
||||
// seatPlans is the seat's plan by key, as an assignment's value reads.
|
||||
seatPlans map[string][]byte
|
||||
// placement is the seat mesh-placement's keys as read, and placed the modules it names on this node.
|
||||
placement *delivery.PlacementKeys
|
||||
placed map[string]bool
|
||||
refused map[string]string // a module whose plan cannot be acted on, and why
|
||||
config map[string]map[string]json.RawMessage
|
||||
applied map[string]bool // module → a report is kept
|
||||
reported map[string][]byte // module → the last report said, undated
|
||||
last map[string]Report // module → the last report said, dated
|
||||
secrets map[string]heldSecret // the vault's name → what this run holds of it; never written anywhere
|
||||
// items is every item recorded for this node, by collection and contributor: what a collector is filled from.
|
||||
items map[string]map[string]itemValue
|
||||
// recorded is each item's key and its value as kept: what this client compares before it records again.
|
||||
@@ -202,7 +205,7 @@ func (c *Client) Run(ctx context.Context) error {
|
||||
ctx, cancel := context.WithCancel(ctx)
|
||||
defer cancel()
|
||||
kvs := map[string]jetstream.KeyValue{}
|
||||
for _, b := range []string{PlansBucket, ControllerAssignmentsBucket, ConfigBucket, RegistryBucket, AppliedBucket,
|
||||
for _, b := range []string{PlansBucket, PlacementBucket, ConfigBucket, RegistryBucket, AppliedBucket,
|
||||
CollectBucket} {
|
||||
kv, err := c.JS.KeyValue(ctx, b)
|
||||
if err != nil {
|
||||
@@ -239,9 +242,10 @@ func (c *Client) Run(ctx context.Context) error {
|
||||
return nil
|
||||
}
|
||||
mine := c.Node + ".>"
|
||||
// The registry whole, and every machine's assignment: a provider here is bound by consumers on any machine, in
|
||||
// their plans (ADR 0373). Every other bucket, this machine's keys alone.
|
||||
for _, b := range []string{RegistryBucket, PlansBucket, ControllerAssignmentsBucket} {
|
||||
// The registry whole, and every machine's plan: a provider here is bound by consumers on any machine, in their plans
|
||||
// (ADR 0373); and placement whole, as its account is granted it (its node key and the legacy key are both read).
|
||||
// Every other bucket, this machine's keys alone.
|
||||
for _, b := range []string{RegistryBucket, PlansBucket, PlacementBucket} {
|
||||
if err := watch(b, ">"); err != nil {
|
||||
return err
|
||||
}
|
||||
@@ -331,12 +335,13 @@ func (c *Client) Run(ctx context.Context) error {
|
||||
config: map[string]map[string]json.RawMessage{}, applied: map[string]bool{},
|
||||
reported: map[string][]byte{}, secrets: map[string]heldSecret{}, items: map[string]map[string]itemValue{},
|
||||
recorded: map[string][]byte{}, assigned: map[string]bool{}, binds: map[string]binding{},
|
||||
placement: delivery.NewPlacementKeys(), placed: map[string]bool{},
|
||||
providing: map[string]bool{}, leftOut: map[string]string{}, bound: map[string]map[string][]byte{},
|
||||
boundReady: map[string]bool{}, unplaced: map[string]bool{}, boundUsers: map[string]map[string]bool{}}
|
||||
// Every current value first: nothing is filled from half of them.
|
||||
current := map[string]bool{}
|
||||
var early []*nats.Msg
|
||||
for len(current) < 6 { // the registry, the plans, the controller's assignments, config, applied, collect
|
||||
for len(current) < 6 { // the registry, the plans, placement, config, applied, collect
|
||||
select {
|
||||
case <-ctx.Done():
|
||||
return ctx.Err()
|
||||
@@ -448,6 +453,8 @@ func (c *Client) Run(ctx context.Context) error {
|
||||
c.report(ctx, kvs, s, Report{Module: m, Status: StatusFailed, Reason: s.refused[m]})
|
||||
case hasPlan(s, m):
|
||||
c.reconcile(ctx, kvs, s, m)
|
||||
case s.placed[m]:
|
||||
// Placed here with no plan ever read: nothing to fill until the seat composes one.
|
||||
default:
|
||||
c.leave(ctx, kvs, s, m)
|
||||
}
|
||||
@@ -512,7 +519,7 @@ func (c *Client) told(u update) {
|
||||
if module, name, ok := strings.Cut(rest, "."); ok {
|
||||
c.tell(module, Changed{Kind: AskConfig, Key: name})
|
||||
}
|
||||
case PlansBucket, ControllerAssignmentsBucket:
|
||||
case PlansBucket:
|
||||
c.tell(rest, Changed{Kind: AskConfig})
|
||||
}
|
||||
}
|
||||
@@ -560,34 +567,30 @@ func (c *Client) take(s *state, u update) []string {
|
||||
}
|
||||
// A kind's answer moves what every module waits on.
|
||||
return sortedKeys(s.plans)
|
||||
case PlansBucket, ControllerAssignmentsBucket:
|
||||
case PlacementBucket:
|
||||
return c.takePlacement(s, key, e.Value(), e.Revision(), gone)
|
||||
case PlansBucket:
|
||||
if s.seatPlans == nil {
|
||||
s.seatPlans, s.controllerAssigned = map[string][]byte{}, map[string][]byte{}
|
||||
}
|
||||
source := s.seatPlans
|
||||
if u.bucket == ControllerAssignmentsBucket {
|
||||
source = s.controllerAssigned
|
||||
s.seatPlans = map[string][]byte{}
|
||||
}
|
||||
if gone {
|
||||
delete(source, key)
|
||||
} else if u.bucket == ControllerAssignmentsBucket {
|
||||
source[key] = e.Value()
|
||||
delete(s.seatPlans, key)
|
||||
} else if v, err := asAssignment(e.Value()); err != nil {
|
||||
logf("[provisioner-client] the plan %s cannot be read, so the controller's assignment is read: %v", key, err)
|
||||
delete(source, key)
|
||||
logf("[provisioner-client] the plan %s cannot be read, so it is taken as gone: %v", key, err)
|
||||
delete(s.seatPlans, key)
|
||||
} else {
|
||||
source[key] = v
|
||||
s.seatPlans[key] = v
|
||||
}
|
||||
// The seat's plan where it holds one, the controller's assignment otherwise: a module leaves only when
|
||||
// neither places it, never because the seat has no plan of it yet.
|
||||
value, placed := s.seatPlans[key]
|
||||
if !placed {
|
||||
value, placed = s.controllerAssigned[key]
|
||||
}
|
||||
gone = !placed
|
||||
value, planned := s.seatPlans[key]
|
||||
gone = !planned
|
||||
// A value that is not a standing assignment (ended, or in a state: ADR 0408 clause 4) is read as a removed
|
||||
// key, so the module leaves the machine as it does when the key goes (issue 1098).
|
||||
gone = gone || !provider.Standing(value)
|
||||
if module, mine := strings.CutPrefix(key, c.Node+"."); mine && gone && s.placed[module] {
|
||||
// Placement still names it: it stays, filled from its last plan and binding as it did, until neither names
|
||||
// it (takePlacement).
|
||||
return []string{module}
|
||||
}
|
||||
// Whom it binds, whichever machine's it is: a provider here listing its consumers is filled again.
|
||||
var providers []string
|
||||
if ended, had := s.binds[key]; gone && had {
|
||||
@@ -695,12 +698,12 @@ func (c *Client) take(s *state, u update) []string {
|
||||
func (c *Client) leftModules(s *state) []string {
|
||||
left := map[string]bool{}
|
||||
for m, kept := range s.applied {
|
||||
if kept && !hasPlan(s, m) && s.refused[m] == "" {
|
||||
if kept && !hasPlan(s, m) && s.refused[m] == "" && !s.placed[m] {
|
||||
left[m] = true
|
||||
}
|
||||
}
|
||||
for key := range s.recorded {
|
||||
if _, m, ok := itemOfKey(c.Node, key); ok && !hasPlan(s, m) && s.refused[m] == "" {
|
||||
if _, m, ok := itemOfKey(c.Node, key); ok && !hasPlan(s, m) && s.refused[m] == "" && !s.placed[m] {
|
||||
left[m] = true
|
||||
}
|
||||
}
|
||||
@@ -1154,6 +1157,56 @@ func (c *Client) forgetItems(ctx context.Context, kvs map[string]jetstream.KeyVa
|
||||
return removed
|
||||
}
|
||||
|
||||
// takePlacement folds one key of the seat mesh-placement's placement in, and answers the modules of this node it moved:
|
||||
// one it names now stays here without a plan; one it no longer names, with no plan of the seat, leaves.
|
||||
func (c *Client) takePlacement(s *state, key string, value []byte, revision uint64, gone bool) []string {
|
||||
node, err := s.placement.Put(key, value, revision, gone)
|
||||
if node != c.Node {
|
||||
return nil
|
||||
}
|
||||
if err != nil {
|
||||
logf("[provisioner-client] the placement's %s cannot be read, so this node's placement stays as it was: %v", key, err)
|
||||
return nil
|
||||
}
|
||||
now := map[string]bool{}
|
||||
if p, _, ok := s.placement.Of(c.Node); ok {
|
||||
for _, m := range p.Modules {
|
||||
now[m.Module] = true
|
||||
}
|
||||
}
|
||||
var moved []string
|
||||
for m := range now {
|
||||
if !s.placed[m] {
|
||||
s.placed[m] = true
|
||||
if !s.assigned[m] {
|
||||
s.assigned[m] = true
|
||||
moved = append(moved, m)
|
||||
}
|
||||
}
|
||||
}
|
||||
for m := range s.placed {
|
||||
if now[m] {
|
||||
continue
|
||||
}
|
||||
delete(s.placed, m)
|
||||
if _, planned := s.seatPlans[c.Node+"."+m]; planned {
|
||||
continue
|
||||
}
|
||||
// Neither names it: it leaves, and binds nothing more.
|
||||
if ended, had := s.binds[c.Node+"."+m]; had {
|
||||
c.forgetBound(s, ended)
|
||||
delete(s.binds, c.Node+"."+m)
|
||||
}
|
||||
delete(s.plans, m)
|
||||
delete(s.refused, m)
|
||||
delete(s.assigned, m)
|
||||
c.forgetHeld(s, m)
|
||||
moved = append(moved, m)
|
||||
}
|
||||
sort.Strings(moved)
|
||||
return moved
|
||||
}
|
||||
|
||||
// asAssignment is a plan of the seat mesh-provisioner as the client's rules read an assignment's value:
|
||||
// `{node, module, provisioning: plan}`.
|
||||
func asAssignment(raw []byte) ([]byte, error) {
|
||||
|
||||
@@ -81,7 +81,7 @@ func startBus(t *testing.T) *aBus {
|
||||
t.Fatal(err)
|
||||
}
|
||||
b := &aBus{url: s.ClientURL(), domain: js, conn: nc, kv: map[string]jetstream.KeyValue{}}
|
||||
for _, name := range []string{PlansBucket, ControllerAssignmentsBucket, ConfigBucket, RegistryBucket, AppliedBucket,
|
||||
for _, name := range []string{PlansBucket, PlacementBucket, ConfigBucket, RegistryBucket, AppliedBucket,
|
||||
CollectBucket, delivery.HaveBucket} {
|
||||
kv, err := js.CreateKeyValue(context.Background(), jetstream.KeyValueConfig{Bucket: name})
|
||||
if err != nil {
|
||||
@@ -97,7 +97,7 @@ func startBus(t *testing.T) *aBus {
|
||||
`"bucket":"mesh-provisioner_collect","key":"<node>.<collection>.<contributor>"}`)
|
||||
}
|
||||
for _, kind := range placementKinds {
|
||||
b.put(t, RegistryBucket, kind, `{"kind":"`+kind+`","domain":"mesh-controller","from":"bucket",`+
|
||||
b.put(t, RegistryBucket, kind, `{"kind":"`+kind+`","domain":"mesh-placement","from":"bucket",`+
|
||||
`"bucket":"mesh-provisioner_plans","key":"<node>.<module>"}`)
|
||||
}
|
||||
return b
|
||||
@@ -122,6 +122,36 @@ func (b *aBus) put(t *testing.T, bucket, key, value string) {
|
||||
}
|
||||
}
|
||||
|
||||
// place writes this node's placement as the seat mesh-placement keeps it, naming exactly these modules.
|
||||
func (b *aBus) place(t *testing.T, modules ...string) {
|
||||
t.Helper()
|
||||
p := delivery.Placement{V: 1, Node: node}
|
||||
for _, m := range modules {
|
||||
p.Modules = append(p.Modules, delivery.PlacedModule{Module: m})
|
||||
}
|
||||
keys := delivery.NewPlacementKeys()
|
||||
ctx := context.Background()
|
||||
have, _ := b.kv[PlacementBucket].Keys(ctx)
|
||||
for _, k := range have {
|
||||
if e, err := b.kv[PlacementBucket].Get(ctx, k); err == nil {
|
||||
_, _ = keys.Put(k, e.Value(), e.Revision(), false)
|
||||
}
|
||||
}
|
||||
writes, err := keys.Writes(node, &p)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
for _, w := range writes {
|
||||
if w.Value == nil {
|
||||
b.del(t, PlacementBucket, w.Key)
|
||||
continue
|
||||
}
|
||||
if _, err := b.kv[PlacementBucket].Put(ctx, w.Key, w.Value); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
func (b *aBus) del(t *testing.T, bucket, key string) {
|
||||
t.Helper()
|
||||
if err := b.kv[bucket].Delete(context.Background(), key); err != nil {
|
||||
@@ -628,38 +658,34 @@ func TestNatsAModuleTakenOffTheNodeLosesItsReport(t *testing.T) {
|
||||
})
|
||||
}
|
||||
|
||||
// Review of 2026-10-30: until the controller hands its lease away, a module the seat holds no plan of is read from
|
||||
// the controller's assignment (nats on novox runs without a want, so the holder writes it no plan), and a module never
|
||||
// leaves because its plan is missing: only when neither source places it.
|
||||
func TestNatsAModuleWithoutASeatPlanIsFilledFromTheControllersAssignmentAndLeavesOnlyWhenNeitherPlacesIt(t *testing.T) {
|
||||
// Issue 1126 B12 (ADR 0590): the controller's assignments are frozen and go, so what places a module here is the seat
|
||||
// mesh-placement's placement alone. A module stays while placement names it, even with no plan of the seat
|
||||
// mesh-provisioner (nats on novox runs without a want): its last plan keeps being filled, and it leaves only when
|
||||
// neither its plan nor its placement is there.
|
||||
func TestNatsAModuleStaysWhilePlacementNamesItAndLeavesOnlyWhenNeitherPlacesIt(t *testing.T) {
|
||||
b := startBus(t)
|
||||
heard := b.listen(t)
|
||||
e := startEngine(t, map[string]string{dunstrc: "owned"})
|
||||
b.config(t, "dunst", "font-size", "13")
|
||||
p, err := json.Marshal(dunst(t))
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
assignment := fmt.Sprintf(`{"node":%q,"module":"dunst","assigned":"2026-10-14T20:00:00Z","provisioning":%s}`, node, p)
|
||||
b.put(t, ControllerAssignmentsBucket, node+".dunst", assignment)
|
||||
b.place(t, "dunst")
|
||||
b.assign(t, "dunst", dunst(t))
|
||||
aClient(t, b, e, t.TempDir())
|
||||
eventually(t, "dunstrc is filled from the controller's assignment", func() bool {
|
||||
eventually(t, "dunstrc is filled from the seat's plan", func() bool {
|
||||
return strings.Contains(e.read(t, dunstrc), "Monospace 13")
|
||||
})
|
||||
|
||||
// The seat's plan comes, and goes again: the module stays, filled, its report kept.
|
||||
b.assign(t, "dunst", dunst(t))
|
||||
// The seat's plan goes, placement stays: the module stays, filled from its last plan, its report kept.
|
||||
b.del(t, PlansBucket, node+".dunst")
|
||||
b.config(t, "dunst", "font-size", "15")
|
||||
eventually(t, "dunstrc follows a change with only the controller's assignment", func() bool {
|
||||
eventually(t, "dunstrc follows a change with only placement naming it", func() bool {
|
||||
return strings.Contains(e.read(t, dunstrc), "Monospace 15")
|
||||
})
|
||||
if r, _ := heard.last("dunst"); r.Status == StatusRemoved || len(b.keys(t, AppliedBucket, node+".dunst")) != 1 {
|
||||
t.Fatalf("the module left when its plan went, though the controller still places it: %+v", r)
|
||||
t.Fatalf("the module left when its plan went, though placement still names it: %+v", r)
|
||||
}
|
||||
|
||||
// Neither places it: it leaves.
|
||||
b.del(t, ControllerAssignmentsBucket, node+".dunst")
|
||||
b.place(t)
|
||||
eventually(t, "prov.applied says it is removed", func() bool {
|
||||
r, _ := heard.last("dunst")
|
||||
return r.Status == StatusRemoved
|
||||
|
||||
+1
-1
@@ -11,7 +11,7 @@ accounts {
|
||||
users = [
|
||||
{ user: "domain", password: "x" }
|
||||
{ user: "provisioner.g14", password: "x", permissions: {
|
||||
publish: { allow: ["$JS.API.CONSUMER.CREATE.KV_mesh-config_resolved.*.$KV.mesh-config_resolved.g14.>", "$JS.API.CONSUMER.CREATE.KV_mesh-provisioner_plans.>", "$JS.API.CONSUMER.CREATE.KV_mesh-controller_assignments.>", "$JS.API.CONSUMER.CREATE.KV_mesh-provisioner_applied.*.$KV.mesh-provisioner_applied.g14.>", "$JS.API.CONSUMER.CREATE.KV_mesh-provisioner_collect.*.$KV.mesh-provisioner_collect.g14.>", "$JS.API.CONSUMER.CREATE.KV_mesh-provisioner_registry.>", "$JS.API.CONSUMER.CREATE.KV_node-engine_have.*.$KV.node-engine_have.g14.node", "$JS.API.CONSUMER.DELETE.KV_node-engine_have.>", "$JS.API.DIRECT.GET.KV_node-engine_have.$KV.node-engine_have.g14.node", "$JS.API.STREAM.INFO.KV_node-engine_have", "$JS.FC.KV_node-engine_have.>", "$JS.API.CONSUMER.CREATE.KV_postgres_bound.*.$KV.postgres_bound.g14.>", "$JS.API.CONSUMER.CREATE.KV_route-adapter_bound.*.$KV.route-adapter_bound.g14.>", "$JS.API.CONSUMER.CREATE.KV_route-proxy_bound.*.$KV.route-proxy_bound.g14.>", "$JS.API.CONSUMER.CREATE.KV_node-tools_bound.*.$KV.node-tools_bound.g14.>", "$JS.API.CONSUMER.DELETE.KV_mesh-config_resolved.>", "$JS.API.CONSUMER.DELETE.KV_mesh-provisioner_plans.>", "$JS.API.CONSUMER.DELETE.KV_mesh-controller_assignments.>", "$JS.API.CONSUMER.DELETE.KV_mesh-provisioner_applied.>", "$JS.API.CONSUMER.DELETE.KV_mesh-provisioner_collect.>", "$JS.API.CONSUMER.DELETE.KV_mesh-provisioner_registry.>", "$JS.API.CONSUMER.DELETE.KV_postgres_bound.>", "$JS.API.CONSUMER.DELETE.KV_route-adapter_bound.>", "$JS.API.CONSUMER.DELETE.KV_route-proxy_bound.>", "$JS.API.CONSUMER.DELETE.KV_node-tools_bound.>", "$JS.API.DIRECT.GET.KV_mesh-config_resolved.$KV.mesh-config_resolved.g14.>", "$JS.API.DIRECT.GET.KV_mesh-provisioner_plans.>", "$JS.API.DIRECT.GET.KV_mesh-controller_assignments.>", "$JS.API.DIRECT.GET.KV_mesh-provisioner_applied.$KV.mesh-provisioner_applied.g14.>", "$JS.API.DIRECT.GET.KV_mesh-provisioner_collect.$KV.mesh-provisioner_collect.g14.>", "$JS.API.DIRECT.GET.KV_mesh-provisioner_registry.>", "$JS.API.DIRECT.GET.KV_postgres_bound.$KV.postgres_bound.g14.>", "$JS.API.DIRECT.GET.KV_route-adapter_bound.$KV.route-adapter_bound.g14.>", "$JS.API.DIRECT.GET.KV_route-proxy_bound.$KV.route-proxy_bound.g14.>", "$JS.API.DIRECT.GET.KV_node-tools_bound.$KV.node-tools_bound.g14.>", "$JS.API.STREAM.INFO.KV_mesh-config_resolved", "$JS.API.STREAM.INFO.KV_mesh-provisioner_plans", "$JS.API.STREAM.INFO.KV_mesh-controller_assignments", "$JS.API.STREAM.INFO.KV_mesh-provisioner_applied", "$JS.API.STREAM.INFO.KV_mesh-provisioner_collect", "$JS.API.STREAM.INFO.KV_mesh-provisioner_registry", "$JS.API.STREAM.INFO.KV_postgres_bound", "$JS.API.STREAM.INFO.KV_route-adapter_bound", "$JS.API.STREAM.INFO.KV_route-proxy_bound", "$JS.API.STREAM.INFO.KV_node-tools_bound", "$JS.FC.KV_mesh-config_resolved.>", "$JS.FC.KV_mesh-provisioner_plans.>", "$JS.FC.KV_mesh-controller_assignments.>", "$JS.FC.KV_mesh-provisioner_applied.>", "$JS.FC.KV_mesh-provisioner_collect.>", "$JS.FC.KV_mesh-provisioner_registry.>", "$JS.FC.KV_postgres_bound.>", "$JS.FC.KV_route-adapter_bound.>", "$JS.FC.KV_route-proxy_bound.>", "$JS.FC.KV_node-tools_bound.>", "$KV.mesh-provisioner_applied.g14.>", "$KV.mesh-provisioner_collect.g14.>", "mesh.seat.mesh-vault.call.get.provisioner~g14", "prov.applied.g14.>", "prov.changed.g14.>"] }
|
||||
publish: { allow: ["$JS.API.CONSUMER.CREATE.KV_mesh-config_resolved.*.$KV.mesh-config_resolved.g14.>", "$JS.API.CONSUMER.CREATE.KV_mesh-provisioner_plans.>", "$JS.API.CONSUMER.CREATE.KV_mesh-delivery-assigner_placement.>", "$JS.API.CONSUMER.CREATE.KV_mesh-provisioner_applied.*.$KV.mesh-provisioner_applied.g14.>", "$JS.API.CONSUMER.CREATE.KV_mesh-provisioner_collect.*.$KV.mesh-provisioner_collect.g14.>", "$JS.API.CONSUMER.CREATE.KV_mesh-provisioner_registry.>", "$JS.API.CONSUMER.CREATE.KV_node-engine_have.*.$KV.node-engine_have.g14.node", "$JS.API.CONSUMER.DELETE.KV_node-engine_have.>", "$JS.API.DIRECT.GET.KV_node-engine_have.$KV.node-engine_have.g14.node", "$JS.API.STREAM.INFO.KV_node-engine_have", "$JS.FC.KV_node-engine_have.>", "$JS.API.CONSUMER.CREATE.KV_postgres_bound.*.$KV.postgres_bound.g14.>", "$JS.API.CONSUMER.CREATE.KV_route-adapter_bound.*.$KV.route-adapter_bound.g14.>", "$JS.API.CONSUMER.CREATE.KV_route-proxy_bound.*.$KV.route-proxy_bound.g14.>", "$JS.API.CONSUMER.CREATE.KV_node-tools_bound.*.$KV.node-tools_bound.g14.>", "$JS.API.CONSUMER.DELETE.KV_mesh-config_resolved.>", "$JS.API.CONSUMER.DELETE.KV_mesh-provisioner_plans.>", "$JS.API.CONSUMER.DELETE.KV_mesh-delivery-assigner_placement.>", "$JS.API.CONSUMER.DELETE.KV_mesh-provisioner_applied.>", "$JS.API.CONSUMER.DELETE.KV_mesh-provisioner_collect.>", "$JS.API.CONSUMER.DELETE.KV_mesh-provisioner_registry.>", "$JS.API.CONSUMER.DELETE.KV_postgres_bound.>", "$JS.API.CONSUMER.DELETE.KV_route-adapter_bound.>", "$JS.API.CONSUMER.DELETE.KV_route-proxy_bound.>", "$JS.API.CONSUMER.DELETE.KV_node-tools_bound.>", "$JS.API.DIRECT.GET.KV_mesh-config_resolved.$KV.mesh-config_resolved.g14.>", "$JS.API.DIRECT.GET.KV_mesh-provisioner_plans.>", "$JS.API.DIRECT.GET.KV_mesh-delivery-assigner_placement.>", "$JS.API.DIRECT.GET.KV_mesh-provisioner_applied.$KV.mesh-provisioner_applied.g14.>", "$JS.API.DIRECT.GET.KV_mesh-provisioner_collect.$KV.mesh-provisioner_collect.g14.>", "$JS.API.DIRECT.GET.KV_mesh-provisioner_registry.>", "$JS.API.DIRECT.GET.KV_postgres_bound.$KV.postgres_bound.g14.>", "$JS.API.DIRECT.GET.KV_route-adapter_bound.$KV.route-adapter_bound.g14.>", "$JS.API.DIRECT.GET.KV_route-proxy_bound.$KV.route-proxy_bound.g14.>", "$JS.API.DIRECT.GET.KV_node-tools_bound.$KV.node-tools_bound.g14.>", "$JS.API.STREAM.INFO.KV_mesh-config_resolved", "$JS.API.STREAM.INFO.KV_mesh-provisioner_plans", "$JS.API.STREAM.INFO.KV_mesh-delivery-assigner_placement", "$JS.API.STREAM.INFO.KV_mesh-provisioner_applied", "$JS.API.STREAM.INFO.KV_mesh-provisioner_collect", "$JS.API.STREAM.INFO.KV_mesh-provisioner_registry", "$JS.API.STREAM.INFO.KV_postgres_bound", "$JS.API.STREAM.INFO.KV_route-adapter_bound", "$JS.API.STREAM.INFO.KV_route-proxy_bound", "$JS.API.STREAM.INFO.KV_node-tools_bound", "$JS.FC.KV_mesh-config_resolved.>", "$JS.FC.KV_mesh-provisioner_plans.>", "$JS.FC.KV_mesh-delivery-assigner_placement.>", "$JS.FC.KV_mesh-provisioner_applied.>", "$JS.FC.KV_mesh-provisioner_collect.>", "$JS.FC.KV_mesh-provisioner_registry.>", "$JS.FC.KV_postgres_bound.>", "$JS.FC.KV_route-adapter_bound.>", "$JS.FC.KV_route-proxy_bound.>", "$JS.FC.KV_node-tools_bound.>", "$KV.mesh-provisioner_applied.g14.>", "$KV.mesh-provisioner_collect.g14.>", "mesh.seat.mesh-vault.call.get.provisioner~g14", "prov.applied.g14.>", "prov.changed.g14.>"] }
|
||||
subscribe: { allow: ["_INBOX.provisioner.g14.>", "mesh.mod.mesh-vault.event.secret-changed", "prov.ask.g14"] }
|
||||
allow_responses: { max: 1, ttl: "1m" }
|
||||
} }
|
||||
|
||||
Reference in New Issue
Block a user