jschoubben
7c78fd2e0c
mesh/delivery closed: a newer head of the same pull request
mesh/check failed (60): integration: mesh-knowledge/cmd/mesh-knowledge.TestARecordThatNeedsTheOperatorIsAc…; and 59 more | 7c78fd2e0c4f on f469eb387a30
A pass runs on every change of its node, and a newly placed module stays unpublished for the minutes its build takes, so counting passes made every new placement unhealthy while nothing was wrong. The health now says a hold only once a module has stayed unpublished at one build for longer than 45 minutes (the build step's bound is 30), with since when; a published module or another build starts the clock again. Claude-Session: https://claude.ai/code/session_01NcbgfJ9Gum47zkFKWRaovL
1622 lines
57 KiB
Go
1622 lines
57 KiB
Go
package main
|
|
|
|
// The seat's holder (novox/hq ADR 0570 clauses 1 to 5, 7, as changed by ADR 0593 §8; issue 1954 H1-H3): mesh-provisioner
|
|
// makes every placed module's provisioning plan and every provision from bus state, on change events and never on a
|
|
// timer, and keeps them in its states `plans` (`<node>.<module>`) and `provisions`
|
|
// (`<node>.<consumer>.<provision>[.<local>]`), which it alone writes. It composes nothing: each plan is the one the
|
|
// node-engine's published composition carries (`composed-<node>`, ADR 0593 §6), the node-engine alone composing.
|
|
//
|
|
// - **Made per node.** A change of a node's placement, of one of its wants or of one of its published compositions
|
|
// makes that node's plans again; a change of the indexer's module does it for the nodes the module is placed on.
|
|
// Each node's plans are read twice: at the builds its wants name (its plans, what its client applies) and at the
|
|
// newest build of each placed module, so a build that adds a requirement has its key before its want (clause 2).
|
|
// A provision is needed while either needs it (clause 5).
|
|
// - **A key's state comes from its provider's outcome alone** (clause 3, 7): the provider's own entry for that
|
|
// consumer, `<provider>_bound` under the same key: present is provided (the vault's generation carried into the
|
|
// reference, clause 6), refused is failed with why, waits is waits with what. The holder never sets provided by
|
|
// itself: a key of a provider that writes no outcome stays asked.
|
|
// - **Retiring is asked** (clause 5): a key no longer needed is `retiring`, one approve ask under mesh-provisioner's
|
|
// name says what goes from which provider on which node; the warrant goes into the key and the provider acts on it;
|
|
// a decline is kept and not asked again while nothing changes; a key needed again returns to its outcome's state.
|
|
// - **Retiring is judged from a whole picture only** (issue 940, met before as 937 and 1332): no pass judges a
|
|
// retirement before every watched read has loaded once, nor while a module on the node has no composition published
|
|
// at its build; a key whose consumer still has a want or a have on the node, and no plan read in this pass, is left
|
|
// as it is; a pass that would retire most of a node's consumers at once retires nothing and says it as a fault of
|
|
// the mesh (provisioner_health), never as a question; at most askOpenAtMost asks stand open at once, and an ask the
|
|
// router does not hold is raised again.
|
|
|
|
import (
|
|
"bytes"
|
|
"context"
|
|
"encoding/json"
|
|
"errors"
|
|
"fmt"
|
|
"slices"
|
|
"sort"
|
|
"strings"
|
|
"sync"
|
|
"time"
|
|
|
|
stdio "git.novox.be/novox/mesh-sdk/go"
|
|
"git.novox.be/novox/mesh-sdk/go/asks"
|
|
"git.novox.be/novox/mesh-sdk/go/catalogue"
|
|
"git.novox.be/novox/mesh-sdk/go/delivery"
|
|
"git.novox.be/novox/mesh-sdk/go/provisions"
|
|
)
|
|
|
|
// The states it keeps, beside Registry, Needs, Applied and Collect. RefusedState holds, by `<node>.<module>`, why a
|
|
// placed module has no plan composed (issue 774): written while it is held, deleted once it composes again.
|
|
const (
|
|
ProvisionsState = provisions.ProvisionsState
|
|
PlansState = provisions.PlansState
|
|
RefusedState = "refused"
|
|
)
|
|
|
|
// Refusal is one module held on a node, as RefusedState keeps it.
|
|
type Refusal struct {
|
|
Node string `json:"node"`
|
|
Module string `json:"module"`
|
|
Why string `json:"why"`
|
|
Since time.Time `json:"since"`
|
|
}
|
|
|
|
// The reads it makes the plans from. ComposedRead is every node's published compositions, `composed-<node>` (ADR 0593
|
|
// §6), read as the node-engine's named state: its keys are `<node>.<module>.<fingerprint hex>` and `<node>.node`.
|
|
const (
|
|
PlacementRead = delivery.PlacementRead
|
|
WantsRead = delivery.AssignerModule + ".wants"
|
|
ModulesRead = delivery.IndexerModule + ".modules"
|
|
ManifestsRead = delivery.IndexerModule + ".manifests"
|
|
HaveRead = "node-engine.have"
|
|
ComposedRead = "mesh-host.composed"
|
|
)
|
|
|
|
// BoundReads are the providers' entries this module's manifest reads, `<module>.bound`, whose changes it hears. Which
|
|
// path judges a provider is its manifest's word (keepsEntries), not this list: a provider that declares `bound` and is
|
|
// not read here leaves its keys asked, saying its entries are not read. route-proxy (mesh-control) and node-tools
|
|
// (mesh-tools) answer route and mcp-endpoint by the provider helper from outside this catalogue (issue 774, sweep of
|
|
// 2026-10-30 20:10: their keys stood asked on every node).
|
|
var BoundReads = []string{"gitea", "keycloak", "postgres", "route-adapter", "route-proxy", "node-tools", "mesh-vault"}
|
|
|
|
// boundRead is a provider's entries as this module reads them.
|
|
func boundRead(module string) string { return module + ".bound" }
|
|
|
|
// Composer is a node's provisioning plans: at the builds keptAt names (module → fingerprint, the newest build of every
|
|
// other placed module), and at the newest build of each placed module. A module whose composition is refused, or not
|
|
// published yet, is held (issue 774, one bad module must not drop its node): in held with why, none of its plan in
|
|
// plans; a scope's Err is a failure that is no single module's.
|
|
type Composer func(ctx context.Context, node string, keptAt map[string]string) Composition
|
|
|
|
// Composition is a node's plans at its wanted builds and at its newest, and where the time went.
|
|
type Composition struct {
|
|
AtWants, Newest HeldPlans
|
|
// Unpublished is each module whose composition at the build it is read at is not published yet: the node's picture
|
|
// is not whole, so no retirement is judged on it (issue 940).
|
|
Unpublished []string
|
|
Took []Phase
|
|
}
|
|
|
|
// HeldPlans is one scope's plans by module, the modules held with why, and a failure that is no single module's.
|
|
type HeldPlans struct {
|
|
Plans map[string]catalogue.Plan
|
|
Held map[string]string
|
|
Err error
|
|
}
|
|
|
|
// Phase is one step of reading a node's plans and how long it took.
|
|
type Phase struct {
|
|
Name string
|
|
Took time.Duration
|
|
}
|
|
|
|
// Asker is the operator's channel as this module asks on it: publish an ask under its own name, and read the router's
|
|
// record of it.
|
|
type Asker interface {
|
|
Publish(subject string, body any, id string) error
|
|
Record(id string) ([]byte, bool, error)
|
|
}
|
|
|
|
// Holder is the seat's holder.
|
|
type Holder struct {
|
|
Store Store
|
|
Compose Composer
|
|
Asks Asker
|
|
Now func() time.Time
|
|
Logf func(format string, args ...any)
|
|
// Loaded says which watched reads have not loaded once yet, "" when all have (nil: all have). A pass before then
|
|
// judges nothing (issue 940: a first read that was not whole retired 21 routes of running apps).
|
|
Loaded func() string
|
|
|
|
mu sync.Mutex
|
|
// faults is, by node, the retirement a pass held back as a fault of the mesh (issue 940), said by provisioner_health.
|
|
faults map[string]string
|
|
// unpublished is, by node and module, since when the module is held because its composition at this build is not
|
|
// published, with the hold's words: one held longer than unpublishedFor is said by provisioner_health.
|
|
unpublished map[string]map[string]unpublishedHold
|
|
}
|
|
|
|
// Composed is what one node's composition did.
|
|
type Composed struct {
|
|
Node string `json:"node"`
|
|
Plans []string `json:"plans-written,omitempty"`
|
|
Gone []string `json:"plans-removed,omitempty"`
|
|
Changed []string `json:"provisions-changed,omitempty"`
|
|
Asked []string `json:"retiring-asked,omitempty"`
|
|
// Held is each module placed on the node whose plan is not composed, with why: its last plan and its provisions
|
|
// stay as they were, and every other module is composed (issue 774).
|
|
Held map[string]string `json:"modules-held,omitempty"`
|
|
Problems []string `json:"problems,omitempty"`
|
|
// Took is where the composition's time went, phase by phase in milliseconds, its whole last (issue 1954).
|
|
Took []Took `json:"took"`
|
|
}
|
|
|
|
// Took is one phase of a composition and its milliseconds.
|
|
type Took struct {
|
|
Phase string `json:"phase"`
|
|
MS int64 `json:"ms"`
|
|
}
|
|
|
|
// TookSaid says a composition's phases in one line.
|
|
func TookSaid(took []Took) string {
|
|
var said []string
|
|
for _, t := range took {
|
|
said = append(said, fmt.Sprintf("%s %d ms", t.Phase, t.MS))
|
|
}
|
|
return strings.Join(said, ", ")
|
|
}
|
|
|
|
// ComposeNode composes one node: its plans and its provisions.
|
|
func (h *Holder) ComposeNode(ctx context.Context, node string) (Composed, error) {
|
|
return h.ComposeNodeWith(ctx, node, false)
|
|
}
|
|
|
|
// ComposeNodeWith composes one node; retireMany lets this one pass retire more of the node's consumers at once than
|
|
// the bound (manyAtOnce) allows, each still asked: the operator's word that a large removal is meant.
|
|
func (h *Holder) ComposeNodeWith(ctx context.Context, node string, retireMany bool) (out Composed, err error) {
|
|
h.mu.Lock()
|
|
defer h.mu.Unlock()
|
|
out = Composed{Node: node}
|
|
began, start := time.Now(), time.Now()
|
|
mark := func(phase string, took time.Duration) {
|
|
out.Took = append(out.Took, Took{Phase: phase, MS: took.Milliseconds()})
|
|
}
|
|
lap := func(phase string) {
|
|
now := time.Now()
|
|
mark(phase, now.Sub(start))
|
|
start = now
|
|
}
|
|
defer func() { mark("whole", time.Since(began)) }()
|
|
if h.Loaded != nil {
|
|
if missing := h.Loaded(); missing != "" {
|
|
return out, fmt.Errorf("%s is not composed: %s has not loaded once yet, and a pass over a part judges nothing", node, missing)
|
|
}
|
|
}
|
|
wanted, err := h.wantedOn(node)
|
|
if err != nil {
|
|
return out, err
|
|
}
|
|
running, err := h.runningOn(node)
|
|
if err != nil {
|
|
return out, err
|
|
}
|
|
// present is every module with a want or a have on the node: one of them with no plan read in this pass is not
|
|
// unneeded (issue 940).
|
|
present := map[string]bool{}
|
|
for module := range wanted {
|
|
present[module] = true
|
|
}
|
|
// A module held on the node with no want (the bus on the control-node) is composed at the build its have says it
|
|
// runs, so it has a plan too and its client keeps filling it (review of 2026-10-30).
|
|
for module, fp := range running {
|
|
present[module] = true
|
|
if _, ok := wanted[module]; !ok {
|
|
wanted[module] = fp
|
|
}
|
|
}
|
|
lap("wants")
|
|
c := h.Compose(ctx, node, wanted)
|
|
for _, p := range c.Took {
|
|
mark(p.Name, p.Took)
|
|
}
|
|
start = time.Now()
|
|
if c.AtWants.Err != nil {
|
|
return out, fmt.Errorf("the plans of %s cannot be composed: %w", node, c.AtWants.Err)
|
|
}
|
|
if c.Newest.Err != nil {
|
|
return out, fmt.Errorf("the plans of %s at the newest builds cannot be composed: %w", node, c.Newest.Err)
|
|
}
|
|
atWants, heldAtWants, newest, heldNewest := c.AtWants.Plans, c.AtWants.Held, c.Newest.Plans, c.Newest.Held
|
|
// A module held at its wanted build has no plan composed: its last plan and its provisions are kept, said in
|
|
// `refused`. One held at its newest build alone keeps its plan at its want; what its newest build needs ahead waits.
|
|
if len(heldAtWants) > 0 {
|
|
out.Held = heldAtWants
|
|
}
|
|
for _, m := range sortedKeys(heldNewest) {
|
|
if _, ok := heldAtWants[m]; !ok {
|
|
out.Problems = append(out.Problems, fmt.Sprintf("%s at its newest build is not composed: %s", m, heldNewest[m]))
|
|
}
|
|
}
|
|
if err := h.writeRefused(node, heldAtWants); err != nil {
|
|
out.Problems = append(out.Problems, err.Error())
|
|
}
|
|
// A module whose composition is not published at its build is held (heldAtWants): its own plan and keys are kept,
|
|
// and nothing else on the node waits for it (review 1 of issue 940). A hold that lasts is said on the health.
|
|
h.countUnpublished(node, c.Unpublished, heldAtWants)
|
|
if err := h.writePlans(node, atWants, heldAtWants, &out); err != nil {
|
|
out.Problems = append(out.Problems, err.Error())
|
|
}
|
|
lap("plans written")
|
|
needed := map[string]provisions.Provision{}
|
|
elsewhere := map[string]provisions.Provision{}
|
|
// ahead are the keys only the newest build of their consumer requires: the plan written, its wanted build's, names
|
|
// them not yet, and the provider binds them from the entry they carry (issue 1126).
|
|
ahead := map[string]string{}
|
|
vaultNode := h.vaultNode()
|
|
for i, plans := range []map[string]catalogue.Plan{atWants, newest} {
|
|
for module, plan := range plans {
|
|
for _, p := range plan.Provisions {
|
|
consumer := provisions.End{Module: module, Node: node}
|
|
provider := provisions.End{Module: p.ProviderModule, Node: p.Provider}
|
|
pv := provisions.Provision{Provision: p.Provision, Local: p.Local, Consumer: consumer, Provider: provider,
|
|
DependsOn: provisions.DependsOn(consumer, p, h.providerPlan(p, node, atWants))}
|
|
key, err := pv.Key()
|
|
if err != nil {
|
|
out.Problems = append(out.Problems, err.Error())
|
|
continue
|
|
}
|
|
if _, seen := needed[key]; !seen {
|
|
if i == 1 {
|
|
// A key no plan names carries the newest build's entry, so its provider binds it (issue 1126). Only
|
|
// it: a reader on an older mesh-sdk refuses a field it does not know, and skips then only a key it
|
|
// could never bind anyway.
|
|
entry := p
|
|
pv.Plan = &entry
|
|
ahead[key] = module
|
|
}
|
|
needed[key] = pv
|
|
}
|
|
// Its credential is a provision of its own, the vault's (clause 6): on the consumer's node for a value of the
|
|
// consumer's own, on the provider's for the provider's one shared value.
|
|
ck, ok := provisions.CredentialKey(p.CredentialForm, consumer, provider, p.Provision, p.Local)
|
|
if !ok {
|
|
continue
|
|
}
|
|
cn, cm, cp, cl, err := provisions.ParseKey(ck)
|
|
if err != nil {
|
|
continue
|
|
}
|
|
if vaultNode == "" {
|
|
out.Problems = append(out.Problems, "no node runs mesh-vault, so the credential "+ck+" has no provider")
|
|
continue
|
|
}
|
|
cred := provisions.Provision{Provision: cp, Local: cl, Consumer: provisions.End{Module: cm, Node: cn},
|
|
Provider: provisions.End{Module: provisions.VaultModule, Node: vaultNode}}
|
|
if cn == node {
|
|
if _, seen := needed[ck]; !seen {
|
|
needed[ck] = cred
|
|
}
|
|
} else {
|
|
elsewhere[ck] = cred
|
|
}
|
|
}
|
|
}
|
|
}
|
|
// A module with a want or a have on the node whose plan this pass did not read is left as it is: its keys are not
|
|
// judged unneeded (issue 940).
|
|
untouched := map[string]string{}
|
|
for m, why := range heldAtWants {
|
|
untouched[m] = why
|
|
}
|
|
for m := range present {
|
|
_, a := atWants[m]
|
|
_, n := newest[m]
|
|
if _, u := untouched[m]; !a && !n && !u {
|
|
untouched[m] = "it has a want or a have on " + node + " and no plan of it was read"
|
|
}
|
|
}
|
|
h.judgeElsewhere(elsewhere, &out)
|
|
h.judgeNode(node, needed, ahead, untouched, retireMany, &out)
|
|
lap("provisions judged")
|
|
return out, nil
|
|
}
|
|
|
|
// writeRefused keeps, for each module held on node, why, and deletes the refusal of one that composes again: a change
|
|
// of the state is the event a watcher hears.
|
|
func (h *Holder) writeRefused(node string, held map[string]string) error {
|
|
var problems []error
|
|
for _, module := range sortedKeys(held) {
|
|
key, err := provisions.PlanKey(node, module)
|
|
if err != nil {
|
|
problems = append(problems, err)
|
|
continue
|
|
}
|
|
r := Refusal{Node: node, Module: module, Why: held[module], Since: h.Now().UTC()}
|
|
if e, err := h.Store.Get(RefusedState, key); err == nil && e != nil {
|
|
var was Refusal
|
|
if json.Unmarshal(e.Value, &was) == nil && was.Why == r.Why {
|
|
continue
|
|
}
|
|
}
|
|
raw, err := json.Marshal(r)
|
|
if err != nil {
|
|
problems = append(problems, err)
|
|
continue
|
|
}
|
|
if err := h.Store.Put(RefusedState, key, json.RawMessage(raw)); err != nil {
|
|
problems = append(problems, fmt.Errorf("the refusal %s: %w", key, err))
|
|
continue
|
|
}
|
|
h.Logf("[%s] %s on %s has no plan composed: %s", Module, module, node, r.Why)
|
|
}
|
|
keys, err := h.Store.Keys(RefusedState)
|
|
if err != nil {
|
|
return errors.Join(append(problems, err)...)
|
|
}
|
|
for _, k := range keys {
|
|
n, module, ok := strings.Cut(k, ".")
|
|
if !ok || n != node {
|
|
continue
|
|
}
|
|
if _, still := held[module]; still {
|
|
continue
|
|
}
|
|
if err := h.Store.Delete(RefusedState, k); err != nil {
|
|
problems = append(problems, err)
|
|
continue
|
|
}
|
|
h.Logf("[%s] %s on %s composes again", Module, module, node)
|
|
}
|
|
return errors.Join(problems...)
|
|
}
|
|
|
|
// providerPlan is the plan of a provision's provider on its node: from this composition when it is on this node, else
|
|
// as kept in `plans`; nil when the provider is not handed.
|
|
func (h *Holder) providerPlan(p catalogue.PlanProvision, node string, here map[string]catalogue.Plan) *catalogue.Plan {
|
|
if p.Provider == node {
|
|
if plan, ok := here[p.ProviderModule]; ok {
|
|
return &plan
|
|
}
|
|
return nil
|
|
}
|
|
key, err := provisions.PlanKey(p.Provider, p.ProviderModule)
|
|
if err != nil {
|
|
return nil
|
|
}
|
|
e, err := h.Store.Get(PlansState, key)
|
|
if err != nil || e == nil {
|
|
return nil
|
|
}
|
|
plan, err := provisions.DecodePlan(e.Value)
|
|
if err != nil {
|
|
return nil
|
|
}
|
|
return &plan.Plan
|
|
}
|
|
|
|
// vaultNode is the node mesh-vault is placed on: the provider of every credential.
|
|
func (h *Holder) vaultNode() string {
|
|
nodes := h.placedOn(provisions.VaultModule)
|
|
if len(nodes) == 0 {
|
|
return ""
|
|
}
|
|
return nodes[0]
|
|
}
|
|
|
|
// dependedOn is every key of a node that a provision of another node depends on.
|
|
func (h *Holder) dependedOn(node string) []string {
|
|
var out []string
|
|
keys, err := h.Store.Keys(ProvisionsState)
|
|
if err != nil {
|
|
return nil
|
|
}
|
|
for _, k := range keys {
|
|
if strings.HasPrefix(k, node+".") {
|
|
continue
|
|
}
|
|
e, err := h.Store.Get(ProvisionsState, k)
|
|
if err != nil || e == nil {
|
|
continue
|
|
}
|
|
if p, err := provisions.Decode(e.Value); err == nil {
|
|
for _, d := range p.DependsOn {
|
|
if strings.HasPrefix(d, node+".") {
|
|
out = append(out, d)
|
|
}
|
|
}
|
|
}
|
|
}
|
|
return out
|
|
}
|
|
|
|
// judgeElsewhere writes each needed key of another node (a provider's shared credential), its state from its outcome;
|
|
// that node's own composition keeps it while a provision depends on it.
|
|
func (h *Holder) judgeElsewhere(keys map[string]provisions.Provision, out *Composed) {
|
|
for _, key := range sortedKeys(keys) {
|
|
next := keys[key]
|
|
var have provisions.Provision
|
|
e, err := h.Store.Get(ProvisionsState, key)
|
|
if err == nil && e != nil {
|
|
if have, err = provisions.Decode(e.Value); err == nil {
|
|
next.Reference = have.Reference
|
|
}
|
|
}
|
|
h.outcome(key, &next)
|
|
if e != nil && same(have, next) {
|
|
continue
|
|
}
|
|
h.put(key, next)
|
|
out.Changed = append(out.Changed, key+" "+next.State)
|
|
}
|
|
}
|
|
|
|
// wantedOn is the fingerprint each want on a node names, by module.
|
|
func (h *Holder) wantedOn(node string) (map[string]string, error) {
|
|
out := map[string]string{}
|
|
keys, err := h.Store.Keys(WantsRead)
|
|
if err != nil {
|
|
return nil, fmt.Errorf("the wants cannot be read: %w", err)
|
|
}
|
|
for _, k := range keys {
|
|
n, module, err := delivery.ParseWantKey(k)
|
|
if err != nil || n != node {
|
|
continue
|
|
}
|
|
e, err := h.Store.Get(WantsRead, k)
|
|
if err != nil || e == nil {
|
|
continue
|
|
}
|
|
if w, err := delivery.DecodeWant(e.Value); err == nil && !w.Removed && w.FP != "" {
|
|
out[module] = w.FP
|
|
}
|
|
}
|
|
return out, nil
|
|
}
|
|
|
|
// runningOn is the fingerprint each module's have on a node says it runs, by module: what a module held without a want
|
|
// is composed at.
|
|
func (h *Holder) runningOn(node string) (map[string]string, error) {
|
|
out := map[string]string{}
|
|
keys, err := h.Store.Keys(HaveRead)
|
|
if err != nil {
|
|
return nil, fmt.Errorf("the haves cannot be read: %w", err)
|
|
}
|
|
for _, k := range keys {
|
|
n, module, isNode, err := delivery.ParseHaveKey(k)
|
|
if err != nil || isNode || n != node {
|
|
continue
|
|
}
|
|
e, err := h.Store.Get(HaveRead, k)
|
|
if err != nil || e == nil {
|
|
continue
|
|
}
|
|
if have, err := delivery.DecodeHave(e.Value); err == nil && have.FP != "" {
|
|
out[module] = have.FP
|
|
}
|
|
}
|
|
return out, nil
|
|
}
|
|
|
|
// writePlans writes each plan that differs and removes the node's plans of modules no longer composed there; a held
|
|
// module's plan is kept as it was (issue 774).
|
|
func (h *Holder) writePlans(node string, plans map[string]catalogue.Plan, held map[string]string, out *Composed) error {
|
|
var problems []error
|
|
for _, module := range sortedKeys(plans) {
|
|
key, err := provisions.PlanKey(node, module)
|
|
if err != nil {
|
|
problems = append(problems, err)
|
|
continue
|
|
}
|
|
raw, err := provisions.EncodePlan(provisions.Plan{Node: node, Module: module, Plan: plans[module]})
|
|
if err != nil {
|
|
problems = append(problems, err)
|
|
continue
|
|
}
|
|
if e, err := h.Store.Get(PlansState, key); err == nil && e != nil && bytes.Equal(e.Value, raw) {
|
|
continue
|
|
}
|
|
if err := h.Store.Put(PlansState, key, json.RawMessage(raw)); err != nil {
|
|
problems = append(problems, fmt.Errorf("the plan %s: %w", key, err))
|
|
continue
|
|
}
|
|
out.Plans = append(out.Plans, key)
|
|
}
|
|
keys, err := h.Store.Keys(PlansState)
|
|
if err != nil {
|
|
return errors.Join(append(problems, err)...)
|
|
}
|
|
for _, k := range keys {
|
|
n, module, ok := strings.Cut(k, ".")
|
|
if !ok || n != node {
|
|
continue
|
|
}
|
|
if _, still := plans[module]; still {
|
|
continue
|
|
}
|
|
if _, kept := held[module]; kept {
|
|
continue
|
|
}
|
|
if err := h.Store.Delete(PlansState, k); err != nil {
|
|
problems = append(problems, err)
|
|
continue
|
|
}
|
|
out.Gone = append(out.Gone, k)
|
|
}
|
|
return errors.Join(problems...)
|
|
}
|
|
|
|
// judgeNode brings the node's provision keys to what is needed and what each provider said. A key of a module left
|
|
// untouched (held on the node, or with a want or a have and no plan read in this pass) that its composition no longer
|
|
// names is left as it is: not composed is not unneeded. A pass that would retire most of the node's consumers at once retires none, said as a fault of
|
|
// the mesh (issue 940), unless retireMany.
|
|
func (h *Holder) judgeNode(node string, needed map[string]provisions.Provision, ahead map[string]string,
|
|
untouched map[string]string, retireMany bool, out *Composed) {
|
|
held := h.held(node)
|
|
consumers := map[string]bool{}
|
|
for key, have := range held {
|
|
consumers[have.Consumer.Module] = true
|
|
if _, isNeeded := needed[key]; !isNeeded && have.Consumer.Node == node {
|
|
if _, left := untouched[have.Consumer.Module]; left {
|
|
delete(held, key)
|
|
}
|
|
}
|
|
}
|
|
// A key of this node another node's provision depends on is needed (a provider's one shared credential).
|
|
for _, dep := range h.dependedOn(node) {
|
|
if have, ok := held[dep]; ok {
|
|
if _, n := needed[dep]; !n {
|
|
have.State, have.Why, have.Ask, have.Warrant = "", "", "", nil
|
|
needed[dep] = have
|
|
}
|
|
}
|
|
}
|
|
// The keys this pass would turn retiring, and whose they are.
|
|
var newly []string
|
|
leaving := map[string]bool{}
|
|
for key, have := range held {
|
|
if _, isNeeded := needed[key]; !isNeeded && have.State != provisions.Retiring {
|
|
newly = append(newly, key)
|
|
leaving[have.Consumer.Module] = true
|
|
}
|
|
}
|
|
sort.Strings(newly)
|
|
hold := func(why string) {
|
|
for _, key := range newly {
|
|
delete(held, key)
|
|
}
|
|
out.Problems = append(out.Problems, why)
|
|
}
|
|
|
|
switch {
|
|
case len(newly) == 0:
|
|
h.clearFault(node)
|
|
case manyAtOnce(len(leaving), len(consumers)) && !retireMany:
|
|
fault := fmt.Sprintf("On %s, one pass would retire the provisions of %d of the %d consumers that hold any (%s): "+
|
|
"most likely a read stopped listing them (a node's compositions published in part, a want, a have or the "+
|
|
"placement not read whole), not their removal. Nothing is retired and nothing is asked; the keys stay as they "+
|
|
"were. If the removal is meant, provisioner_compose with retire-many retires them, each asked.",
|
|
node, len(leaving), len(consumers), strings.Join(sortedKeys(leaving), ", "))
|
|
hold(fault)
|
|
h.setFault(node, fault)
|
|
default:
|
|
h.clearFault(node)
|
|
}
|
|
for _, key := range unionKeys(needed, held) {
|
|
want, isNeeded := needed[key]
|
|
have, isHeld := held[key]
|
|
var next provisions.Provision
|
|
switch {
|
|
case isNeeded:
|
|
next = want
|
|
if isHeld {
|
|
next.Reference = have.Reference
|
|
}
|
|
h.outcome(key, &next)
|
|
if module, isAhead := ahead[key]; isAhead && next.State == provisions.Asked {
|
|
// The reason the provider's helper says too: no plan names it yet, and it binds it from this key's entry.
|
|
why := "no plan names it yet: the newest build of " + module + " requires it and its wanted build does " +
|
|
"not, so its provider binds it from this key's plan entry"
|
|
if next.Why != "" {
|
|
why += "; " + next.Why
|
|
}
|
|
next.Why = why
|
|
}
|
|
case have.State == provisions.Retiring:
|
|
next = have
|
|
if next.Ask != "" && next.Warrant == nil && !h.askHeld(next) {
|
|
out.Problems = append(out.Problems, "the router holds no ask "+next.Ask+" of "+key+": it is raised again")
|
|
next.Ask = ""
|
|
}
|
|
if next.Ask == "" && next.Warrant == nil {
|
|
h.raise(&next, key, out)
|
|
}
|
|
if gone, why := h.retired(key, have); gone {
|
|
if err := h.Store.Delete(ProvisionsState, key); err != nil {
|
|
out.Problems = append(out.Problems, err.Error())
|
|
} else {
|
|
out.Changed = append(out.Changed, key+" (retired: "+why+")")
|
|
}
|
|
continue
|
|
}
|
|
default:
|
|
next = have
|
|
next.State, next.Why = provisions.Retiring, fmt.Sprintf("no build wanted or placed on %s requires %s of %s on %s "+
|
|
"any more", node, have.Provision, have.Provider.Module, have.Provider.Node)
|
|
next.Warrant = nil
|
|
h.raise(&next, key, out)
|
|
}
|
|
if isHeld && same(have, next) {
|
|
continue
|
|
}
|
|
next.At = h.Now().UTC()
|
|
raw, err := provisions.Encode(next)
|
|
if err != nil {
|
|
out.Problems = append(out.Problems, err.Error())
|
|
continue
|
|
}
|
|
if err := h.Store.Put(ProvisionsState, key, json.RawMessage(raw)); err != nil {
|
|
out.Problems = append(out.Problems, fmt.Sprintf("%s: %v", key, err))
|
|
continue
|
|
}
|
|
out.Changed = append(out.Changed, key+" "+next.State)
|
|
}
|
|
}
|
|
|
|
// held is the node's provision keys as kept.
|
|
func (h *Holder) held(node string) map[string]provisions.Provision {
|
|
out := map[string]provisions.Provision{}
|
|
keys, err := h.Store.Keys(ProvisionsState)
|
|
if err != nil {
|
|
return out
|
|
}
|
|
for _, k := range keys {
|
|
if !strings.HasPrefix(k, node+".") {
|
|
continue
|
|
}
|
|
e, err := h.Store.Get(ProvisionsState, k)
|
|
if err != nil || e == nil {
|
|
continue
|
|
}
|
|
if p, err := provisions.Decode(e.Value); err == nil {
|
|
out[k] = p
|
|
}
|
|
}
|
|
return out
|
|
}
|
|
|
|
// same says two provisions say the same, their time aside.
|
|
func same(a, b provisions.Provision) bool {
|
|
a.At, b.At, a.V, b.V = time.Time{}, time.Time{}, 0, 0
|
|
ra, _ := json.Marshal(a)
|
|
rb, _ := json.Marshal(b)
|
|
return bytes.Equal(ra, rb)
|
|
}
|
|
|
|
// entry is a provider's outcome for one consumer, as its entry says it: the provider helper's Entry, read for what
|
|
// the holder needs and nothing more.
|
|
type entry struct {
|
|
// State and Why are the provider's outcome (provider.Entry, ADR 0570 clause 3): provided, failed or waits.
|
|
State string `json:"state,omitempty"`
|
|
Why string `json:"why,omitempty"`
|
|
Refused string `json:"refused,omitempty"`
|
|
Waits string `json:"waits,omitempty"`
|
|
VaultName string `json:"vault-name,omitempty"`
|
|
Generation int `json:"generation,omitempty"`
|
|
}
|
|
|
|
// outcome sets a needed provision's state from its provider's entry: provided, failed or waits; asked while it has
|
|
// none. The holder never sets provided by itself (clause 3): a provider whose entries it does not read keeps the key
|
|
// asked, said so.
|
|
func (h *Holder) outcome(key string, p *provisions.Provision) {
|
|
p.Ask, p.Warrant = "", nil
|
|
keeps, err := h.keepsEntries(p.Provider.Module, p.Provision)
|
|
if err != nil {
|
|
p.State, p.Why = provisions.Asked, "the manifest of "+p.Provider.Module+" cannot be read: "+err.Error()
|
|
return
|
|
}
|
|
if !keeps {
|
|
h.haveOutcome(p)
|
|
return
|
|
}
|
|
if !slices.Contains(BoundReads, p.Provider.Module) {
|
|
p.State, p.Why = provisions.Asked, p.Provider.Module+" keeps its outcome in its entries, which mesh-provisioner's "+
|
|
"manifest does not read yet"
|
|
return
|
|
}
|
|
e, err := h.Store.Get(boundRead(p.Provider.Module), key)
|
|
if err != nil {
|
|
p.State, p.Why = provisions.Asked, "the provider's entries cannot be read: "+err.Error()
|
|
return
|
|
}
|
|
if e == nil {
|
|
p.State, p.Why = provisions.Asked, h.unread(p.Provider.Module)
|
|
return
|
|
}
|
|
var en entry
|
|
if err := json.Unmarshal(e.Value, &en); err != nil {
|
|
p.State, p.Why = provisions.Asked, "the provider's entry is not readable"
|
|
return
|
|
}
|
|
switch {
|
|
case en.State == provisions.Failed:
|
|
p.State, p.Why = provisions.Failed, orSaid(en.Why, en.Refused, "the provider said it failed")
|
|
case en.State == provisions.Waits:
|
|
p.State, p.Why = provisions.Waits, orSaid(en.Why, en.Waits, "the provider waits for a person")
|
|
case en.State == "" && en.Refused != "":
|
|
p.State, p.Why = provisions.Failed, en.Refused
|
|
case en.State == "" && en.Waits != "":
|
|
p.State, p.Why = provisions.Waits, en.Waits
|
|
case en.State != "" && en.State != provisions.Provided:
|
|
p.State, p.Why = provisions.Asked, "the provider's entry says "+en.State+", which no state of a provision is"
|
|
return
|
|
default:
|
|
p.State, p.Why = provisions.Provided, ""
|
|
if en.VaultName != "" || en.Generation > 0 {
|
|
p.Reference = &provisions.Reference{ID: en.VaultName, Generation: en.Generation}
|
|
}
|
|
}
|
|
}
|
|
|
|
// unread is why a provider whose helper keeps its outcome in its entries has written none, when its manifest says it:
|
|
// it reads no plans of the seat (novox/hq ADR 0570 clause 2), so its helper never hears a bind and the key stays asked
|
|
// for good (issue 774, 2026-10-31: route-proxy's helper read only the controller's assignments, and 38 routes the seat
|
|
// keyed for modules not handed stood asked with no why while the assigner withheld their consumers' builds). "" when
|
|
// it reads them, or its manifest cannot say: then it has not answered yet.
|
|
func (h *Holder) unread(module string) string {
|
|
manifest, err := h.providerManifest(module)
|
|
if err != nil {
|
|
return ""
|
|
}
|
|
var decl struct {
|
|
Reads []string `json:"reads"`
|
|
}
|
|
if json.Unmarshal([]byte(manifest), &decl) != nil || slices.Contains(decl.Reads, provisions.PlansRead) {
|
|
return ""
|
|
}
|
|
return module + " reads no plans of the seat (its manifest's reads do not name " + provisions.PlansRead +
|
|
"), so its helper never answers this key"
|
|
}
|
|
|
|
// haveOutcome is the outcome of a provider that makes no credential per consumer: its own have on its node (the lead's
|
|
// reading of ADR 0570 clause 3, recorded as an insight). Running with health ok at the build its want names is
|
|
// provided; failed is failed with the have's words; anything else is asked, waiting for the provider to run.
|
|
func (h *Holder) haveOutcome(p *provisions.Provision) {
|
|
where := p.Provider.Module + " on " + p.Provider.Node
|
|
// It makes what the consumer needs only once all it depends on is provided (its credential first).
|
|
for _, dep := range p.DependsOn {
|
|
state := "not composed yet"
|
|
if e, err := h.Store.Get(ProvisionsState, dep); err == nil && e != nil {
|
|
if d, err := provisions.Decode(e.Value); err == nil {
|
|
state = d.State
|
|
}
|
|
}
|
|
if state != provisions.Provided {
|
|
p.State, p.Why = provisions.Asked, "waits for "+dep+" ("+state+") before "+where+" can provide it"
|
|
return
|
|
}
|
|
}
|
|
key := p.Provider.Node + "." + p.Provider.Module
|
|
e, err := h.Store.Get(HaveRead, key)
|
|
if err != nil {
|
|
p.State, p.Why = provisions.Asked, "the have of "+where+" cannot be read: "+err.Error()
|
|
return
|
|
}
|
|
if e == nil {
|
|
p.State, p.Why = provisions.Asked, "waits for "+where+" to run: it says nothing yet"
|
|
return
|
|
}
|
|
have, err := delivery.DecodeHave(e.Value)
|
|
if err != nil {
|
|
p.State, p.Why = provisions.Asked, "the have of "+where+" is not readable"
|
|
return
|
|
}
|
|
wantFP := ""
|
|
if we, err := h.Store.Get(WantsRead, "want."+key); err == nil && we != nil {
|
|
if w, err := delivery.DecodeWant(we.Value); err == nil && !w.Removed {
|
|
wantFP = w.FP
|
|
}
|
|
}
|
|
switch {
|
|
case have.Phase == delivery.PhaseFailed || have.Health == delivery.HealthFailed:
|
|
p.State, p.Why = provisions.Failed, orSaid(have.Detail, where+" failed")
|
|
case have.Phase == delivery.PhaseRunning && have.Health == delivery.HealthOK && (wantFP == "" || have.FP == wantFP):
|
|
p.State, p.Why = provisions.Provided, ""
|
|
default:
|
|
p.State, p.Why = provisions.Asked, fmt.Sprintf("waits for %s to run healthy at the build its want names (it is %s, "+
|
|
"health %s)", where, have.Phase, have.Health)
|
|
}
|
|
}
|
|
|
|
// keepsEntries says whether a provider's outcome for one provision is its entries (its manifest at the trunk declares the
|
|
// state `bound`) rather than its own have. When any offer of the provider says `answered-by: helper`, only those offers
|
|
// are answered by its entries: a module running the helper for one provision and making another some other way (gitea:
|
|
// the helper answers forge-account, the vault issues gitea-read, the package harness makes npm-package-registry) is
|
|
// judged per offer, so the helper's "does not serve" for an offer it does not run never becomes the provision's outcome
|
|
// (issue 774, sweep of 2026-10-30 20:10). A provider whose offers say nothing keeps every outcome in its entries.
|
|
func (h *Holder) keepsEntries(module, provision string) (bool, error) {
|
|
man, err := h.providerManifest(module)
|
|
if err != nil {
|
|
return false, err
|
|
}
|
|
return keepsEntriesFor(man, provision)
|
|
}
|
|
|
|
// keepsEntriesFor is keepsEntries over a provider's manifest JSON.
|
|
func keepsEntriesFor(manifest, provision string) (bool, error) {
|
|
var decl struct {
|
|
State []json.RawMessage `json:"state"`
|
|
Provides []json.RawMessage `json:"provides"`
|
|
}
|
|
if err := json.Unmarshal([]byte(manifest), &decl); err != nil {
|
|
return false, err
|
|
}
|
|
bound := false
|
|
for _, raw := range decl.State {
|
|
var name string
|
|
if json.Unmarshal(raw, &name) != nil {
|
|
var full struct {
|
|
Name string `json:"name"`
|
|
}
|
|
_ = json.Unmarshal(raw, &full)
|
|
name = full.Name
|
|
}
|
|
if name == "bound" {
|
|
bound = true
|
|
}
|
|
}
|
|
if !bound {
|
|
return false, nil
|
|
}
|
|
helped, says := false, false
|
|
for _, raw := range decl.Provides {
|
|
var full struct {
|
|
Name string `json:"name"`
|
|
AnsweredBy string `json:"answered-by"`
|
|
}
|
|
if json.Unmarshal(raw, &full) != nil || full.AnsweredBy != catalogue.AnsweredByHelper {
|
|
continue
|
|
}
|
|
says = true
|
|
if full.Name == provision {
|
|
helped = true
|
|
}
|
|
}
|
|
return helped || !says, nil
|
|
}
|
|
|
|
// providerManifest is a provider's manifest JSON at the trunk, as the indexer holds it.
|
|
func (h *Holder) providerManifest(module string) (string, error) {
|
|
mk, err := delivery.ModulesKey(module)
|
|
if err != nil {
|
|
return "", err
|
|
}
|
|
e, err := h.Store.Get(ModulesRead, mk)
|
|
if err != nil || e == nil {
|
|
return "", fmt.Errorf("the indexer holds no modules.%s (%v)", module, err)
|
|
}
|
|
m, err := delivery.DecodeModule(e.Value)
|
|
if err != nil || m.Manifest == "" {
|
|
return "", fmt.Errorf("the indexer's modules.%s names no manifest", module)
|
|
}
|
|
key, err := delivery.ManifestsKey(m.Manifest)
|
|
if err != nil {
|
|
return "", err
|
|
}
|
|
me, err := h.Store.Get(ManifestsRead, key)
|
|
if err != nil || me == nil {
|
|
return "", fmt.Errorf("the indexer holds no manifest of %s (%v)", module, err)
|
|
}
|
|
man, err := delivery.DecodeManifest(me.Value)
|
|
if err != nil {
|
|
return "", err
|
|
}
|
|
return man.JSON, nil
|
|
}
|
|
|
|
// retired says a retiring key the operator approved is retired: its provider's entry is gone.
|
|
func (h *Holder) retired(key string, p provisions.Provision) (bool, string) {
|
|
if p.Warrant == nil || p.Warrant.Option != OptionApprove {
|
|
return false, ""
|
|
}
|
|
if keeps, err := h.keepsEntries(p.Provider.Module, p.Provision); err != nil || !keeps {
|
|
return false, ""
|
|
}
|
|
e, err := h.Store.Get(boundRead(p.Provider.Module), key)
|
|
if err != nil || e != nil {
|
|
return false, ""
|
|
}
|
|
return true, "its provider removed its credential on the operator's warrant"
|
|
}
|
|
|
|
// ---- the retiring ask ----------------------------------------------------------------------------------------------
|
|
|
|
// The options a retiring ask offers. The chosen one is the warrant's option in the key, and a provider lets its entry go
|
|
// only on "yes" (mesh-sdk provider.go, the pass's retiring keys), so the approving option is "yes" (review of
|
|
// 2026-10-30: the holder wrote "approve", which no provider acted on).
|
|
const (
|
|
OptionApprove = "yes"
|
|
OptionDecline = "no"
|
|
)
|
|
|
|
// askLasts is how long a retiring ask stays open.
|
|
const askLasts = asks.ApproveLasts - 10*time.Minute
|
|
|
|
// retiringAsk is the ask of one retiring provision, composed from the key alone.
|
|
func retiringAsk(key string, p provisions.Provision, at time.Time, id string) asks.Ask {
|
|
what := fmt.Sprintf("%s's access to %s on %s, for %s on %s", p.Consumer.Module, p.Provision, p.Provider.Node,
|
|
p.Consumer.Module, p.Consumer.Node)
|
|
a := asks.Ask{ID: id, Headline: "Remove " + p.Consumer.Module + "'s " + p.Provision + " access?",
|
|
Explanation: fmt.Sprintf("Nothing wanted or placed on %s needs %s any more. On Approve, %s removes %s: what it owns "+
|
|
"is handed to the resource's owner and its credential goes; the resource and its data stay.", p.Consumer.Node,
|
|
p.Provision, p.Provider.Module, what),
|
|
Who: asks.Operator, Expires: at.Add(askLasts), OnExpiry: "nothing is removed; it is asked again when it changes",
|
|
About: "provision." + key, Details: "The provision " + key + " of mesh-provisioner's state provisions."}
|
|
for _, o := range []struct{ id, label, does string }{
|
|
{OptionApprove, "Remove", "the provider removes the credential; the resource and its data stay"},
|
|
{OptionDecline, "Keep", "nothing is removed, and it is not asked again while nothing changes"},
|
|
} {
|
|
binds, _ := asks.ActDigest(retireAct(key, id, o.id))
|
|
a.Options = append(a.Options, asks.Option{ID: o.id, Label: o.label, Does: o.does, Level: asks.Approve, Binds: binds})
|
|
}
|
|
return a
|
|
}
|
|
|
|
func retireAct(key, id, option string) asks.Act {
|
|
return asks.Act{"verb": "mesh-provisioner.retire", "provision": key, "ask": id, "option": option}
|
|
}
|
|
|
|
// askOpenAtMost is how many of its asks mesh-provisioner keeps open at once: the router shows at most three per asker
|
|
// and refuses the rest (issue 940: 18 of 21 retiring asks were refused so, yet counted asked). The next is raised as
|
|
// one ends (Decide).
|
|
const askOpenAtMost = 3
|
|
|
|
// askHeldWithin is how long the router may take to hold an ask before an ask it holds no record of is raised again.
|
|
const askHeldWithin = time.Minute
|
|
|
|
// askWaits is what a retiring key's why says while its ask waits for an open one to end.
|
|
const askWaits = "its ask waits until fewer than 3 of mesh-provisioner's asks stand open, the router showing at most 3 per asker"
|
|
|
|
// errAskWaits is an ask held back by askOpenAtMost: no fault, raised once one ends.
|
|
var errAskWaits = errors.New(askWaits)
|
|
|
|
// raise asks a retiring key and says so in out: asked, waiting for an open ask to end, or a problem.
|
|
func (h *Holder) raise(p *provisions.Provision, key string, out *Composed) {
|
|
switch err := h.ask(p, key); {
|
|
case err == nil:
|
|
out.Asked = append(out.Asked, key)
|
|
case errors.Is(err, errAskWaits):
|
|
default:
|
|
out.Problems = append(out.Problems, err.Error())
|
|
}
|
|
}
|
|
|
|
// ask raises the retiring ask of a key, under mesh-provisioner's own name; one already raised stands. While
|
|
// askOpenAtMost asks stand open it raises none and the key's why says so.
|
|
func (h *Holder) ask(p *provisions.Provision, key string) error {
|
|
if p.Ask != "" {
|
|
return nil
|
|
}
|
|
if h.Asks == nil {
|
|
return errors.New("the operator cannot be asked from here, so " + key + " is retiring and nothing is removed")
|
|
}
|
|
if h.openAsks() >= askOpenAtMost {
|
|
if !strings.Contains(p.Why, askWaits) {
|
|
p.Why = strings.TrimPrefix(p.Why+"; "+askWaits, "; ")
|
|
}
|
|
return errAskWaits
|
|
}
|
|
now := h.Now().UTC()
|
|
id := "retire-" + strings.ReplaceAll(key, ".", "-") + "-" + fmt.Sprint(now.Unix())
|
|
a := retiringAsk(key, *p, now, id)
|
|
if err := a.Check(now); err != nil {
|
|
return err
|
|
}
|
|
if err := h.Asks.Publish(asks.AskSubject(Module), a, id); err != nil {
|
|
return fmt.Errorf("the retiring ask of %s could not be published: %w", key, err)
|
|
}
|
|
p.Ask = id
|
|
p.Why = strings.TrimSuffix(strings.TrimSuffix(p.Why, askWaits), "; ")
|
|
return nil
|
|
}
|
|
|
|
// openAsks is how many retiring keys hold an ask with no answer yet.
|
|
func (h *Holder) openAsks() int {
|
|
n := 0
|
|
keys, _ := h.Store.Keys(ProvisionsState)
|
|
for _, k := range keys {
|
|
if e, err := h.Store.Get(ProvisionsState, k); err == nil && e != nil {
|
|
if p, err := provisions.Decode(e.Value); err == nil && p.State == provisions.Retiring && p.Ask != "" &&
|
|
p.Warrant == nil {
|
|
n++
|
|
}
|
|
}
|
|
}
|
|
return n
|
|
}
|
|
|
|
// askHeld says the router holds a key's ask, or may not have taken it yet: false only when it has no record of it
|
|
// askHeldWithin after it was raised (issue 940: an ask the router refused was counted asked for good).
|
|
func (h *Holder) askHeld(p provisions.Provision) bool {
|
|
if h.Asks == nil || p.Ask == "" {
|
|
return true
|
|
}
|
|
if _, found, err := h.Asks.Record(p.Ask); err != nil || found {
|
|
return true
|
|
}
|
|
return h.Now().Sub(p.At) < askHeldWithin
|
|
}
|
|
|
|
// Decide reads the router's record of each retiring key's ask and writes the warrant into the key (clause 5): the
|
|
// provider acts on an Approve; a Decline is kept, and not asked again while nothing changes.
|
|
func (h *Holder) Decide() []string {
|
|
h.mu.Lock()
|
|
defer h.mu.Unlock()
|
|
if h.Asks == nil {
|
|
return nil
|
|
}
|
|
keys, err := h.Store.Keys(ProvisionsState)
|
|
if err != nil {
|
|
return []string{"the provisions cannot be read: " + err.Error()}
|
|
}
|
|
var said []string
|
|
for _, k := range keys {
|
|
e, err := h.Store.Get(ProvisionsState, k)
|
|
if err != nil || e == nil {
|
|
continue
|
|
}
|
|
p, err := provisions.Decode(e.Value)
|
|
if err != nil || p.State != provisions.Retiring || p.Ask == "" || p.Warrant != nil {
|
|
continue
|
|
}
|
|
raw, found, err := h.Asks.Record(p.Ask)
|
|
if err == nil && !found && h.Now().Sub(p.At) >= askHeldWithin {
|
|
// The router holds no record of it: not asked, raised again below once an open ask allows.
|
|
p.Ask = ""
|
|
p.Why = strings.TrimPrefix(p.Why+"; "+askWaits, "; ")
|
|
h.put(k, p)
|
|
said = append(said, k+": the router holds no ask of it; raised again")
|
|
continue
|
|
}
|
|
if err != nil || !found {
|
|
continue
|
|
}
|
|
var rec struct {
|
|
State string `json:"state"`
|
|
Ask *asks.Ask `json:"ask"`
|
|
Warrant *asks.Warrant `json:"warrant"`
|
|
}
|
|
if json.Unmarshal(raw, &rec) != nil || rec.State == "open" || rec.Warrant == nil || rec.Ask == nil {
|
|
continue
|
|
}
|
|
w := *rec.Warrant
|
|
if w.Asker != Module || w.Ask != p.Ask {
|
|
continue
|
|
}
|
|
option := ""
|
|
if w.Outcome == asks.OutcomeChosen {
|
|
o, err := w.For(Module, *rec.Ask)
|
|
if err == nil {
|
|
err = o.Performs(retireAct(k, p.Ask, o.ID))
|
|
}
|
|
if err != nil {
|
|
said = append(said, k+": the answer is refused: "+err.Error())
|
|
continue
|
|
}
|
|
option = o.ID
|
|
} else {
|
|
// Ended with no choice: asked again at the next change.
|
|
p.Ask = ""
|
|
h.put(k, p)
|
|
said = append(said, k+": the ask "+string(w.Outcome)+"; asked again when it changes")
|
|
continue
|
|
}
|
|
by := "the operator"
|
|
if w.By != nil {
|
|
by = fmt.Sprintf("the %s, as %s identity %s", w.By.Who, w.By.Kind, w.By.Identity)
|
|
}
|
|
p.Warrant = &provisions.Warrant{Ask: p.Ask, Option: option, By: by, At: w.At.UTC()}
|
|
h.put(k, p)
|
|
said = append(said, k+": "+option+" by "+by)
|
|
}
|
|
return append(said, h.raiseWaiting()...)
|
|
}
|
|
|
|
// raiseWaiting raises the asks that wait for an open one to end, as many as askOpenAtMost allows: an answer, which
|
|
// wakes Decide, is what frees a place.
|
|
func (h *Holder) raiseWaiting() []string {
|
|
keys, err := h.Store.Keys(ProvisionsState)
|
|
if err != nil {
|
|
return nil
|
|
}
|
|
var said []string
|
|
for _, k := range keys {
|
|
e, err := h.Store.Get(ProvisionsState, k)
|
|
if err != nil || e == nil {
|
|
continue
|
|
}
|
|
p, err := provisions.Decode(e.Value)
|
|
if err != nil || p.State != provisions.Retiring || p.Ask != "" || p.Warrant != nil || !strings.Contains(p.Why, askWaits) {
|
|
continue
|
|
}
|
|
switch err := h.ask(&p, k); {
|
|
case errors.Is(err, errAskWaits):
|
|
return said
|
|
case err != nil:
|
|
said = append(said, k+": "+err.Error())
|
|
default:
|
|
h.put(k, p)
|
|
said = append(said, k+": asked")
|
|
}
|
|
}
|
|
return said
|
|
}
|
|
|
|
// manyAtOnce says a pass retiring the provisions of leaving consumers, of all the consumers holding a key on the node,
|
|
// retires most of them at once: more than 3, or more than half and at least 2 (one consumer leaving is a removal, not
|
|
// a fault).
|
|
func manyAtOnce(leaving, all int) bool {
|
|
return leaving > 3 || (leaving >= 2 && 2*leaving > all)
|
|
}
|
|
|
|
func (h *Holder) setFault(node, why string) {
|
|
if h.faults == nil {
|
|
h.faults = map[string]string{}
|
|
}
|
|
if h.faults[node] != why {
|
|
h.Logf("[%s] %s", Module, why)
|
|
}
|
|
h.faults[node] = why
|
|
}
|
|
|
|
func (h *Holder) clearFault(node string) {
|
|
if _, ok := h.faults[node]; ok {
|
|
h.Logf("[%s] the retirement held back on %s is no longer judged", Module, node)
|
|
delete(h.faults, node)
|
|
}
|
|
}
|
|
|
|
// unpublishedFor is how long a module may be held unpublished at one build before the health says it: longer than a
|
|
// build takes (the build step's bound is 30 minutes), so a newly placed module waiting for its build is no fault. A
|
|
// pass runs on every change of its node, so passes are no measure of it (review 2 of issue 940).
|
|
const unpublishedFor = 45 * time.Minute
|
|
|
|
type unpublishedHold struct {
|
|
since time.Time
|
|
why string // the hold's words, which name the build
|
|
}
|
|
|
|
// countUnpublished keeps, for each module held unpublished on node, since when it is held at that build; a module
|
|
// published again, or held at another build, starts again.
|
|
func (h *Holder) countUnpublished(node string, modules []string, held map[string]string) {
|
|
if h.unpublished == nil {
|
|
h.unpublished = map[string]map[string]unpublishedHold{}
|
|
}
|
|
was := h.unpublished[node]
|
|
now := map[string]unpublishedHold{}
|
|
for _, m := range modules {
|
|
u, ok := was[m]
|
|
if !ok || u.why != held[m] {
|
|
u = unpublishedHold{since: h.Now().UTC(), why: held[m]}
|
|
}
|
|
now[m] = u
|
|
}
|
|
h.unpublished[node] = now
|
|
}
|
|
|
|
// Faults is, by node, each retirement a pass held back as a fault of the mesh.
|
|
func (h *Holder) Faults() map[string]string {
|
|
h.mu.Lock()
|
|
defer h.mu.Unlock()
|
|
out := make(map[string]string, len(h.faults))
|
|
for n, why := range h.faults {
|
|
out[n] = why
|
|
}
|
|
return out
|
|
}
|
|
|
|
// LastingHolds is each module held unpublished at one build for longer than unpublishedFor, as "<node>.<module>" with
|
|
// the hold's words, which name its build, and since when.
|
|
func (h *Holder) LastingHolds() map[string]string {
|
|
h.mu.Lock()
|
|
defer h.mu.Unlock()
|
|
out := map[string]string{}
|
|
now := h.Now()
|
|
for n, mods := range h.unpublished {
|
|
for m, u := range mods {
|
|
if now.Sub(u.since) > unpublishedFor {
|
|
out[n+"."+m] = fmt.Sprintf("%s (held since %s UTC)", u.why, u.since.Format("2006-01-02 15:04"))
|
|
}
|
|
}
|
|
}
|
|
return out
|
|
}
|
|
|
|
func (h *Holder) put(key string, p provisions.Provision) {
|
|
p.At = h.Now().UTC()
|
|
if raw, err := provisions.Encode(p); err == nil {
|
|
if err := h.Store.Put(ProvisionsState, key, json.RawMessage(raw)); err != nil {
|
|
h.Logf("[%s] %s is not written: %v", Module, key, err)
|
|
}
|
|
}
|
|
}
|
|
|
|
// ---- what a change touches --------------------------------------------------------------------------------------
|
|
|
|
// NodesOfPlacement is the node a placement key names, in either form (research 072 row 37).
|
|
func NodesOfPlacement(key string) []string {
|
|
if n, _, _, err := delivery.ParsePlacementKey(key); err == nil {
|
|
return []string{n}
|
|
}
|
|
return nil
|
|
}
|
|
|
|
// NodesOfWant is the node a want key names.
|
|
func NodesOfWant(key string) []string {
|
|
if n, _, err := delivery.ParseWantKey(key); err == nil {
|
|
return []string{n}
|
|
}
|
|
return nil
|
|
}
|
|
|
|
// NodesOfModule is every node a module is placed on, and every node whose plans name it as a provider: what a change
|
|
// of the module's manifest touches.
|
|
func (h *Holder) NodesOfModule(module string) []string {
|
|
nodes := map[string]bool{}
|
|
for _, n := range h.placedOn(module) {
|
|
nodes[n] = true
|
|
}
|
|
if keys, err := h.Store.Keys(ProvisionsState); err == nil {
|
|
for _, k := range keys {
|
|
if e, err := h.Store.Get(ProvisionsState, k); err == nil && e != nil {
|
|
if p, err := provisions.Decode(e.Value); err == nil && p.Provider.Module == module {
|
|
nodes[p.Consumer.Node] = true
|
|
}
|
|
}
|
|
}
|
|
}
|
|
return sortedKeys(nodes)
|
|
}
|
|
|
|
// placedOn is every node a module is placed on, sorted, from the placement's keys in either form
|
|
// (delivery.PlacementsOf; research 072 row 37). A node whose keys do not read whole is read as they stand.
|
|
func (h *Holder) placedOn(module string) []string {
|
|
nodes := map[string]bool{}
|
|
raw := map[string][]byte{}
|
|
if keys, err := h.Store.Keys(PlacementRead); err == nil {
|
|
for _, k := range keys {
|
|
if e, err := h.Store.Get(PlacementRead, k); err == nil && e != nil {
|
|
raw[k] = e.Value
|
|
}
|
|
}
|
|
}
|
|
placements, _ := delivery.PlacementsOf(raw)
|
|
for n, p := range placements {
|
|
for _, m := range p.Modules {
|
|
if m.Module == module {
|
|
nodes[n] = true
|
|
}
|
|
}
|
|
}
|
|
return sortedKeys(nodes)
|
|
}
|
|
|
|
// NodeOfEntry is the consumer's node a provider's entry key names.
|
|
func NodeOfEntry(key string) []string {
|
|
if n, _, _, _, err := provisions.ParseKey(key); err == nil {
|
|
return []string{n}
|
|
}
|
|
return nil
|
|
}
|
|
|
|
func sortedKeys[V any](m map[string]V) []string {
|
|
out := make([]string, 0, len(m))
|
|
for k := range m {
|
|
out = append(out, k)
|
|
}
|
|
sort.Strings(out)
|
|
return out
|
|
}
|
|
|
|
func unionKeys[A, B any](a map[string]A, b map[string]B) []string {
|
|
seen := map[string]bool{}
|
|
for k := range a {
|
|
seen[k] = true
|
|
}
|
|
for k := range b {
|
|
seen[k] = true
|
|
}
|
|
return sortedKeys(seen)
|
|
}
|
|
|
|
// Rejudge sets one key's state again from its provider's outcome: what a change of a provider's entry touches. A
|
|
// retiring key whose provider removed its entry on the operator's warrant goes.
|
|
func (h *Holder) Rejudge(key string) (string, error) {
|
|
h.mu.Lock()
|
|
defer h.mu.Unlock()
|
|
e, err := h.Store.Get(ProvisionsState, key)
|
|
if err != nil || e == nil {
|
|
return "", err
|
|
}
|
|
have, err := provisions.Decode(e.Value)
|
|
if err != nil {
|
|
return "", err
|
|
}
|
|
if have.State == provisions.Retiring {
|
|
if gone, why := h.retired(key, have); gone {
|
|
return key + " retired: " + why, h.Store.Delete(ProvisionsState, key)
|
|
}
|
|
return "", nil
|
|
}
|
|
next := have
|
|
h.outcome(key, &next)
|
|
if same(have, next) {
|
|
return "", nil
|
|
}
|
|
h.put(key, next)
|
|
return key + " " + next.State, nil
|
|
}
|
|
|
|
// RetireNeeds deletes every key of the retired state `needs`, which the clients no longer write (clause 1): run once at
|
|
// the start.
|
|
func (h *Holder) RetireNeeds() (int, error) {
|
|
keys, err := h.Store.Keys(Needs)
|
|
if err != nil {
|
|
return 0, err
|
|
}
|
|
n := 0
|
|
for _, k := range keys {
|
|
if err := h.Store.Delete(Needs, k); err != nil {
|
|
return n, err
|
|
}
|
|
n++
|
|
}
|
|
return n, nil
|
|
}
|
|
|
|
// orSaid is the first of the words that says something.
|
|
func orSaid(words ...string) string {
|
|
for _, w := range words {
|
|
if strings.TrimSpace(w) != "" {
|
|
return w
|
|
}
|
|
}
|
|
return ""
|
|
}
|
|
|
|
// FromCompositions is the holder's composer over the node-engine's published compositions (ADR 0593 §8): it composes
|
|
// nothing, and takes each module's plan from the key of `composed-<node>` at the build it is read at. At the wanted
|
|
// builds a module is read at the fingerprint keptAt names, else at the indexer's newest (FP2); at the newest, each placed
|
|
// module at FP2. A key whose engine refused the module holds it with the engine's words; a key not published yet holds
|
|
// it at its wanted build (its last plan kept, as for a refusal) and leaves it out of the newest, whose build the engine
|
|
// writes once it is done (ADR 0593 §6), and says it unpublished, so the pass judges no retirement on the node (issue
|
|
// 940). A placed module no build of which is known is held too. A plan absent from a key is a module that hands the
|
|
// provisioner nothing.
|
|
func FromCompositions(s Store) Composer {
|
|
return func(_ context.Context, node string, keptAt map[string]string) Composition {
|
|
start := time.Now()
|
|
at, newest := HeldPlans{Plans: map[string]catalogue.Plan{}, Held: map[string]string{}},
|
|
HeldPlans{Plans: map[string]catalogue.Plan{}, Held: map[string]string{}}
|
|
published, err := publishedOn(s, node)
|
|
if err != nil {
|
|
at.Err, newest.Err = err, err
|
|
return Composition{AtWants: at, Newest: newest, Took: []Phase{{Name: "compositions read", Took: time.Since(start)}}}
|
|
}
|
|
placed, fp2, err := placedModules(s, node)
|
|
if err != nil {
|
|
at.Err, newest.Err = err, err
|
|
return Composition{AtWants: at, Newest: newest, Took: []Phase{{Name: "compositions read", Took: time.Since(start)}}}
|
|
}
|
|
atWants := map[string]string{}
|
|
for _, m := range placed {
|
|
atWants[m] = fp2[m]
|
|
}
|
|
for m, fp := range keptAt {
|
|
atWants[m] = fp
|
|
}
|
|
var unpublished []string
|
|
for _, m := range sortedKeys(atWants) {
|
|
fp := atWants[m]
|
|
if fp == "" {
|
|
// Placed, with no want, no have and no newest build the indexer names: no build of it is known, which is
|
|
// not a module handing nothing (issue 940).
|
|
at.Held[m] = fmt.Sprintf("no build of %s on %s is known: no want, no have, and the indexer names no newest build", m, node)
|
|
continue
|
|
}
|
|
c, ok := published[composedAt(node, m, fp)]
|
|
switch {
|
|
case !ok:
|
|
unpublished = append(unpublished, m)
|
|
at.Held[m] = fmt.Sprintf("the composition of %s at %s is not published by the node-engine of %s yet", m, fp, node)
|
|
case c.Refused != "":
|
|
at.Held[m] = c.Refused
|
|
case c.Plan != nil:
|
|
at.Plans[m] = *c.Plan
|
|
}
|
|
}
|
|
for _, m := range placed {
|
|
if fp2[m] == "" {
|
|
continue
|
|
}
|
|
c, ok := published[composedAt(node, m, fp2[m])]
|
|
switch {
|
|
case !ok:
|
|
case c.Refused != "":
|
|
newest.Held[m] = c.Refused
|
|
case c.Plan != nil:
|
|
newest.Plans[m] = *c.Plan
|
|
}
|
|
}
|
|
return Composition{AtWants: at, Newest: newest, Unpublished: unpublished,
|
|
Took: []Phase{{Name: "compositions read", Took: time.Since(start)}}}
|
|
}
|
|
}
|
|
|
|
// publishedPlan is what the holder reads of a published composition (delivery.Composed): its plan, or the engine's
|
|
// refusal. Read leniently, so a field the node-engine adds later is no failure here.
|
|
type publishedPlan struct {
|
|
Plan *catalogue.Plan `json:"plan"`
|
|
Refused string `json:"refused"`
|
|
}
|
|
|
|
// composedAt is a module's key in its node's compositions, or "" for a fingerprint no key can name.
|
|
func composedAt(node, module, fp string) string {
|
|
k, err := delivery.ComposedKey(node, module, fp)
|
|
if err != nil {
|
|
return ""
|
|
}
|
|
return k
|
|
}
|
|
|
|
// publishedOn is every module composition the node's engine publishes, by key.
|
|
func publishedOn(s Store, node string) (map[string]publishedPlan, error) {
|
|
keys, err := s.Keys(ComposedRead)
|
|
if err != nil {
|
|
return nil, fmt.Errorf("the compositions the node-engines publish cannot be read: %w", err)
|
|
}
|
|
out := map[string]publishedPlan{}
|
|
for _, k := range keys {
|
|
n, _, _, isNode, err := delivery.ParseComposedKey(k)
|
|
if err != nil || isNode || n != node {
|
|
continue
|
|
}
|
|
e, err := s.Get(ComposedRead, k)
|
|
if err != nil || e == nil {
|
|
continue
|
|
}
|
|
var c publishedPlan
|
|
if err := json.Unmarshal(e.Value, &c); err != nil {
|
|
c.Refused = fmt.Sprintf("its published composition %s does not read: %v", k, err)
|
|
}
|
|
out[k] = c
|
|
}
|
|
return out, nil
|
|
}
|
|
|
|
// placedModules is every module placed on node, sorted, and the indexer's newest fingerprint (FP2) of each.
|
|
func placedModules(s Store, node string) ([]string, map[string]string, error) {
|
|
raw := map[string][]byte{}
|
|
keys, err := s.Keys(PlacementRead)
|
|
if err != nil {
|
|
return nil, nil, fmt.Errorf("the placement cannot be read: %w", err)
|
|
}
|
|
for _, k := range keys {
|
|
if n, _, _, err := delivery.ParsePlacementKey(k); err != nil || n != node {
|
|
continue
|
|
}
|
|
if e, err := s.Get(PlacementRead, k); err == nil && e != nil {
|
|
raw[k] = e.Value
|
|
}
|
|
}
|
|
placements, _ := delivery.PlacementsOf(raw)
|
|
var placed []string
|
|
fp2 := map[string]string{}
|
|
for _, m := range placements[node].Modules {
|
|
placed = append(placed, m.Module)
|
|
mk, err := delivery.ModulesKey(m.Module)
|
|
if err != nil {
|
|
continue
|
|
}
|
|
if e, err := s.Get(ModulesRead, mk); err == nil && e != nil {
|
|
if mod, err := delivery.DecodeModule(e.Value); err == nil {
|
|
fp2[m.Module] = mod.FP2
|
|
}
|
|
}
|
|
}
|
|
sort.Strings(placed)
|
|
return placed, fp2, nil
|
|
}
|
|
|
|
// WatchedReads are the states the holder reads whole, each from the watched copy (ADR 0580 clause 2): the placement,
|
|
// the wants, the indexer's modules, the haves and the published compositions, and the providers' entries, whose watch
|
|
// is also what rejudges a key.
|
|
var WatchedReads = append([]string{PlacementRead, WantsRead, ModulesRead, HaveRead, ComposedRead}, boundReads()...)
|
|
|
|
// OwnStates are the states the holder alone writes (ADR 0570 clause 1), read from the watched copy too (issue 1954: some
|
|
// 250 to 400 reads by key per composition): each of its own writes is taken into the copy at once (WatchedStates.Wrote),
|
|
// so it reads back what it wrote before the watch hands it back.
|
|
var OwnStates = []string{PlansState, ProvisionsState, RefusedState}
|
|
|
|
func boundReads() []string {
|
|
out := make([]string, 0, len(BoundReads))
|
|
for _, p := range BoundReads {
|
|
out = append(out, boundRead(p))
|
|
}
|
|
return out
|
|
}
|
|
|
|
// Watching is a Store whose WatchedReads and OwnStates come from the copy, its writes of OwnStates taken into the copy as
|
|
// they land, and a manifest read once by its digest; every other state, and every write, from s.
|
|
func Watching(s Store, ws *stdio.WatchedStates) Store { return watchingStore{Store: s, ws: ws} }
|
|
|
|
type watchingStore struct {
|
|
Store
|
|
ws *stdio.WatchedStates
|
|
}
|
|
|
|
// RevisionPutter is a Store that says the revision a put landed at (the bus does), so the copy keeps it exactly.
|
|
type RevisionPutter interface {
|
|
PutRevision(state, key string, value any) (uint64, error)
|
|
}
|
|
|
|
func (w watchingStore) copied(state string) bool {
|
|
return slices.Contains(WatchedReads, state) || slices.Contains(OwnStates, state)
|
|
}
|
|
|
|
func (w watchingStore) Get(state, key string) (*Entry, error) {
|
|
get := w.ws.Get
|
|
switch {
|
|
case state == ManifestsRead:
|
|
get = w.ws.Once // a manifest by its digest never changes: read once, kept (ADR 0580 clause 1)
|
|
case !w.copied(state):
|
|
return w.Store.Get(state, key)
|
|
}
|
|
v, found, err := get(state, key)
|
|
if err != nil || !found {
|
|
return nil, err
|
|
}
|
|
return &Entry{Key: key, Value: v}, nil
|
|
}
|
|
|
|
func (w watchingStore) Keys(state string) ([]string, error) {
|
|
if !w.copied(state) {
|
|
return w.Store.Keys(state)
|
|
}
|
|
return w.ws.Keys(state)
|
|
}
|
|
|
|
func (w watchingStore) Put(state, key string, value any) error {
|
|
if !slices.Contains(OwnStates, state) {
|
|
return w.Store.Put(state, key, value)
|
|
}
|
|
raw, err := json.Marshal(value)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
var revision uint64
|
|
if rp, ok := w.Store.(RevisionPutter); ok {
|
|
revision, err = rp.PutRevision(state, key, json.RawMessage(raw))
|
|
} else {
|
|
err = w.Store.Put(state, key, json.RawMessage(raw))
|
|
}
|
|
if err != nil {
|
|
return err
|
|
}
|
|
return w.ws.Wrote(state, key, raw, revision)
|
|
}
|
|
|
|
func (w watchingStore) Delete(state, key string) error {
|
|
if err := w.Store.Delete(state, key); err != nil || !slices.Contains(OwnStates, state) {
|
|
return err
|
|
}
|
|
return w.ws.Removed(state, key)
|
|
}
|
|
|
|
// NodesOfNodeKey is the node a key that starts with it names: a have's `<node>.<module>` or `<node>.node`, a published
|
|
// composition's `<node>.<module>.<fingerprint hex>` or `<node>.node`. A change of either makes that node's plans
|
|
// again (ADR 0580 clause 2, ADR 0593 §8).
|
|
func NodesOfNodeKey(key string) []string {
|
|
if n, _, ok := strings.Cut(key, "."); ok && n != "" {
|
|
return []string{n}
|
|
}
|
|
return nil
|
|
}
|