jschoubben
333ab9de22
A provider whose build does not offer the provision it was asked for must not be counted as having delivered it; the provisioner reads the build's manifest (the builds read serves only this offer check and makes no plan, ADR 0593 section 8) before it takes the provider's have as the outcome.
1683 lines
60 KiB
Go
1683 lines
60 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"
|
|
// BuildsRead is the builder's builds, `<module>.<fingerprint part>`: the key of the build a provider runs names the
|
|
// digest of the manifest it was built from (novox/hq ADR 0603 clause 5).
|
|
BuildsRead = delivery.BuilderModule + ".builds"
|
|
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):
|
|
// A provider's have counts as its outcome only from a build whose manifest offers the provision (ADR 0603 clause 5).
|
|
// Refused only when that manifest is known and does not offer it: a have with no builder key (an old-form build, an
|
|
// adopted module) or an unreadable manifest keeps the outcome it had.
|
|
if offers, known := h.buildOffers(p.Provider.Module, have.FP, p.Provision); known && !offers {
|
|
p.State, p.Why = provisions.Asked, "waits for "+where+" to run a build that offers "+p.Provision+": the build it runs ("+
|
|
have.FP+") does not"
|
|
return
|
|
}
|
|
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)
|
|
}
|
|
}
|
|
|
|
// buildOffers says whether the build of a module at a fingerprint, as the builder keyed it, was made from a manifest that
|
|
// offers the provision; known is false when the builder holds no such key, the key names no manifest, or the indexer holds
|
|
// no such manifest (the caller then does not refuse).
|
|
func (h *Holder) buildOffers(module, fingerprint, provision string) (offers, known bool) {
|
|
key, err := delivery.BuilderBuildKey(module, fingerprint)
|
|
if err != nil {
|
|
return false, false
|
|
}
|
|
be, err := h.Store.Get(BuildsRead, key)
|
|
if err != nil || be == nil {
|
|
return false, false
|
|
}
|
|
b, err := delivery.DecodeBuilderBuild(be.Value)
|
|
if err != nil || b.Manifest == "" {
|
|
return false, false
|
|
}
|
|
mk, err := delivery.ManifestsKey(b.Manifest)
|
|
if err != nil {
|
|
return false, false
|
|
}
|
|
me, err := h.Store.Get(ManifestsRead, mk)
|
|
if err != nil || me == nil {
|
|
return false, false
|
|
}
|
|
man, err := delivery.DecodeManifest(me.Value)
|
|
if err != nil {
|
|
return false, false
|
|
}
|
|
var decl struct {
|
|
Provides []json.RawMessage `json:"provides"`
|
|
}
|
|
if json.Unmarshal([]byte(man.JSON), &decl) != nil {
|
|
return false, false
|
|
}
|
|
for _, raw := range decl.Provides {
|
|
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 == provision {
|
|
return true, true
|
|
}
|
|
}
|
|
return false, true
|
|
}
|
|
|
|
// 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
|
|
}
|