jschoubben
ca5f74eadb
ADR 0539 clause 7 issues a module's credential and grant only when its build is done. A module built long before and assigned to one more node never has a build done again: on 2026-10-25 at 17:19 mesh-build-node and mesh-build-go were assigned to ace and composed for it, the bus's user list never named ace's accounts, and both runtimes restarted every 5 s (one with no credential, one refused by the bus). The assign act composes the assigned node alone, never the node holding the bus, and writes no grant. So a write of the assignments bucket is the event: when a key comes to stand (new, or ended or pending before), the lease holder issues the module's accounts on that node that are not minted (never re-issued), composes the node holding the bus and writes the grants, through the composer the done keys' take-ins use. A rewrite of a standing assignment does nothing; the start's look settles every standing one, one after another. A step of research 072's grants (memberships and credentials heard from placement).
289 lines
11 KiB
Go
289 lines
11 KiB
Go
package main
|
|
|
|
import (
|
|
"context"
|
|
"errors"
|
|
"fmt"
|
|
"slices"
|
|
"sort"
|
|
"strings"
|
|
"sync"
|
|
"time"
|
|
|
|
sdkdelivery "git.novox.be/novox/mesh-sdk/go/delivery"
|
|
"github.com/novox/mesh-controller/internal/builderkeys"
|
|
"github.com/novox/mesh-controller/internal/conditions"
|
|
"github.com/novox/mesh-controller/internal/inventory"
|
|
)
|
|
|
|
// A new module is registered from its first done key (novox/hq ADR 0501 §1, §6; issue 1851, A5).
|
|
//
|
|
// Nothing asks a new module's first build: the builder builds every fingerprint the indexer writes, a new module's
|
|
// included (ADR 0429 §5). The controller registers a module the catalogue does not hold from the builder's done key of the
|
|
// fingerprint the indexer's trunk value names, under the same rules of who may register as before (ADR 0501 §3): from a
|
|
// repository the catalogue builds another module from, or on the operator's Approve of the merge the trunk value names
|
|
// (a first merge, kept with mesh-delivery's word), or at the terminal (ADR 0501 §4). It is assigned nowhere.
|
|
//
|
|
// **No timer** (ADR 0426): a write of the builder or the indexer wakes it for that module (builderkeys.Watch), a first
|
|
// merge's word wakes it for that repository, and a controller that starts looks at every module once (ADR 0501 §7: a
|
|
// manifest the controller before could not read is registered when the controller that reads it runs).
|
|
|
|
// newModulesMu keeps one take-in of new modules at a time in this process.
|
|
var newModulesMu sync.Mutex
|
|
|
|
// newModuleWait is a new module the controller could not register yet, and why: what the first-builds condition names.
|
|
type newModuleWait struct {
|
|
Module, Repository, Commit, Why string
|
|
// Word is the manifest word this controller does not read, when that is why (ADR 0501 §7).
|
|
Word string
|
|
Since time.Time
|
|
}
|
|
|
|
// newModulesWaiting are the new modules waiting, by module, in this process: a restart looks at every module again.
|
|
var newModulesWaiting = struct {
|
|
sync.Mutex
|
|
by map[string]newModuleWait
|
|
}{by: map[string]newModuleWait{}}
|
|
|
|
// takeNewModules looks at the modules the indexer holds that the catalogue does not, all of them or those named, and
|
|
// registers each whose done key may register it. Each outcome is said once: a module still waiting for the same reason
|
|
// is not said again.
|
|
func takeNewModules(ctx context.Context, open *stores, only ...string) {
|
|
newModulesMu.Lock()
|
|
defer newModulesMu.Unlock()
|
|
keys, closeKeys, err := openBuilderKeys()
|
|
if err != nil {
|
|
fmt.Printf("new modules: the builder's keys cannot be read: %v\n", err)
|
|
return
|
|
}
|
|
defer closeKeys()
|
|
for _, module := range takeNewModulesWith(ctx, open.inventory, keys, only...) {
|
|
// An assignment waiting for it is made now (ADR 0501 §6; issue 1851, A10).
|
|
for _, line := range settlePendingOnKey(ctx, open, module) {
|
|
fmt.Println(line)
|
|
}
|
|
}
|
|
}
|
|
|
|
// takeNewModulesWith is takeNewModules on the buckets handed in; it answers the modules it registered.
|
|
func takeNewModulesWith(ctx context.Context, inv *inventory.Inventory, keys newModuleReads, only ...string) []string {
|
|
entries, err := inv.Catalogued(ctx)
|
|
if err != nil {
|
|
fmt.Printf("new modules: the catalogue cannot be read: %v\n", err)
|
|
// empty-on-error: said above; nothing is registered this time, and the next write of a key looks again
|
|
return nil
|
|
}
|
|
held := map[string]bool{}
|
|
for _, e := range entries {
|
|
held[e.Manifest.Module] = true
|
|
}
|
|
values, unread, err := keys.Modules(ctx)
|
|
if err != nil {
|
|
fmt.Printf("new modules: the indexer's modules cannot be read: %v\n", err)
|
|
// empty-on-error: said above; nothing is registered this time, and the next write of a key looks again
|
|
return nil
|
|
}
|
|
for module, err := range unread {
|
|
if !held[module] && (len(only) == 0 || slices.Contains(only, module)) {
|
|
sayNewModuleWait(newModuleWait{Module: module, Why: "the indexer's value of it cannot be read: " + err.Error()})
|
|
}
|
|
}
|
|
names := make([]string, 0, len(values))
|
|
for module := range values {
|
|
if !held[module] && (len(only) == 0 || slices.Contains(only, module)) {
|
|
names = append(names, module)
|
|
}
|
|
}
|
|
sort.Strings(names)
|
|
var registered []string
|
|
for _, module := range names {
|
|
v := values[module]
|
|
_, _, err := takeInKey(ctx, inv, keys, keyTake{Module: module, Trunk: v, Ask: registrationAsk{Merge: v.Commit}})
|
|
switch {
|
|
case err == nil:
|
|
forgetNewModuleWait(module)
|
|
registered = append(registered, module)
|
|
fmt.Printf("%s is new in %s at %.8s: registered from the builder's done key, assigned nowhere\n", module,
|
|
v.Repository, v.Commit)
|
|
case errors.Is(err, errKeyWaits):
|
|
w := newModuleWait{Module: module, Repository: v.Repository, Commit: v.Commit, Why: err.Error()}
|
|
if word := wordIn(err.Error()); word != "" {
|
|
w.Word = word
|
|
}
|
|
sayNewModuleWait(w)
|
|
default:
|
|
sayNewModuleWait(newModuleWait{Module: module, Repository: v.Repository, Commit: v.Commit,
|
|
Why: "not registered: " + err.Error()})
|
|
}
|
|
}
|
|
return registered
|
|
}
|
|
|
|
// newModuleReads is what the take-in of new modules reads of the builder and the indexer.
|
|
type newModuleReads interface {
|
|
keyReads
|
|
Modules(ctx context.Context) (map[string]sdkdelivery.Module, map[string]error, error)
|
|
}
|
|
|
|
var _ newModuleReads = builderkeys.Reader{}
|
|
|
|
// wordIn is the manifest word a take-in's words say this controller does not read.
|
|
func wordIn(said string) string {
|
|
_, rest, ok := strings.Cut(said, "uses the word \"")
|
|
if !ok {
|
|
return ""
|
|
}
|
|
word, _, _ := strings.Cut(rest, "\"")
|
|
return word
|
|
}
|
|
|
|
// sayNewModuleWait keeps a new module waiting and says it, once for each reason.
|
|
func sayNewModuleWait(w newModuleWait) {
|
|
newModulesWaiting.Lock()
|
|
defer newModulesWaiting.Unlock()
|
|
was, known := newModulesWaiting.by[w.Module]
|
|
if known && was.Why == w.Why && was.Commit == w.Commit {
|
|
return
|
|
}
|
|
w.Since = time.Now().UTC()
|
|
if known && was.Commit == w.Commit {
|
|
w.Since = was.Since
|
|
}
|
|
newModulesWaiting.by[w.Module] = w
|
|
fmt.Printf("%s is new at %.8s and not registered yet: %s\n", w.Module, w.Commit, firstLine(w.Why))
|
|
}
|
|
|
|
func forgetNewModuleWait(module string) {
|
|
newModulesWaiting.Lock()
|
|
defer newModulesWaiting.Unlock()
|
|
delete(newModulesWaiting.by, module)
|
|
}
|
|
|
|
// kindNewModuleWaits is what a new module waiting for this controller's reader of its manifest, or waiting past its
|
|
// bound, raises (ADR 0501 §6, §7: the first-builds condition reads the key, not a build request).
|
|
const kindNewModuleWaits = "new-module-waits"
|
|
|
|
// newModuleWaitBound is how long a new module waits for its key before it is said: the builder builds a fingerprint in
|
|
// minutes. Not a delay: its key's write registers it the moment it is done.
|
|
const newModuleWaitBound = 30 * time.Minute
|
|
|
|
// newModuleObservations names the new modules waiting: one whose manifest uses a word this controller does not read at
|
|
// once (nothing is lost, but the operator sees what waits for which controller), any other past the bound.
|
|
func newModuleObservations(now time.Time) []conditions.Observation {
|
|
newModulesWaiting.Lock()
|
|
defer newModulesWaiting.Unlock()
|
|
var out []conditions.Observation
|
|
for _, w := range newModulesWaiting.by {
|
|
if w.Word == "" && now.Sub(w.Since) <= newModuleWaitBound {
|
|
continue
|
|
}
|
|
head := fmt.Sprintf("%s is not registered: %s", w.Module, firstLine(w.Why))
|
|
resolved := "its key is done and its take-in registers it"
|
|
if w.Word != "" {
|
|
head = fmt.Sprintf("%s waits for a controller that reads the manifest word %q", w.Module, w.Word)
|
|
resolved = "a controller that reads the word runs and registers it from the same key, with no second build"
|
|
}
|
|
out = append(out, conditions.Observation{Scope: conditions.ScopeMerge,
|
|
ID: fmt.Sprintf("%s_%s", w.Module, short(w.Commit)), Kind: kindNewModuleWaits, Severity: conditions.Warning,
|
|
Summary: head,
|
|
Said: fmt.Sprintf("%s at %s waits since %s: %s", w.Module, w.Commit, w.Since.Format(time.RFC3339), w.Why),
|
|
Headline: fmt.Sprintf("The new module %s is not registered yet", w.Module),
|
|
Explanation: fmt.Sprintf("The merge %.8s of %s adds %s. The mesh registers it from the builder's done key of its "+
|
|
"fingerprint, and has not yet: %s", w.Commit, w.Repository, w.Module, w.Why),
|
|
Resolved: resolved,
|
|
})
|
|
}
|
|
sort.Slice(out, func(i, j int) bool { return out[i].ID < out[j].ID })
|
|
return out
|
|
}
|
|
|
|
// watchNewModules wakes the take-in of new modules for each module the builder or the indexer writes, until ctx ends, and
|
|
// looks at every module once now. A bucket not there yet is said: the look at start and a first merge's word still work.
|
|
func watchNewModules(ctx context.Context, open *stores) {
|
|
inBackground(ctx, func() {
|
|
takeNewModules(ctx, open)
|
|
})
|
|
// And, beside it, the start's pass of release 6 and the take-in of each registered module's new done key (novox/hq
|
|
// ADR 0539 clauses 5, 7, 10 and 13; composed_start.go, composed_keys.go): the plans' buckets deleted, every node
|
|
// composed, the replaced controller composed, and from now on each new fingerprint registered, composed and granted.
|
|
inBackground(ctx, func() { composeAtStart(ctx, open) })
|
|
watchDoneKeys(ctx, open)
|
|
// And each assignment that comes to stand issues its node's bus accounts and the module's grant (ADR 0543): a module
|
|
// built long before is assigned to one more node with no build done again (issue 1945, assignment_accounts.go).
|
|
watchAssignmentAccounts(ctx, open)
|
|
watchHavesForJudgement(ctx, open)
|
|
// And the keep set keeps the builds every want names (ADR 0539 clause 1), with no plan left to name them.
|
|
open.inventory.SetWanted(wantedOnTheBus)
|
|
keys, closeKeys, err := openBuilderKeys()
|
|
if err != nil {
|
|
fmt.Printf("new modules: the builder's keys are not watched: %v\n", err)
|
|
return
|
|
}
|
|
if err := builderkeys.Watch(ctx, keys.JS, func(module string) {
|
|
wakeNewModule(ctx, open, module)
|
|
}); err != nil {
|
|
closeKeys()
|
|
fmt.Printf("new modules: the builder's keys are not watched: %v\n", err)
|
|
return
|
|
}
|
|
goServing(ctx, func() {
|
|
<-ctx.Done()
|
|
closeKeys()
|
|
})
|
|
}
|
|
|
|
// wakeNewModule takes module in when the catalogue does not hold it: never on the watch's own goroutine, which must keep
|
|
// draining.
|
|
func wakeNewModule(ctx context.Context, open *stores, module string) {
|
|
inBackground(ctx, func() {
|
|
entries, err := open.inventory.Catalogued(ctx)
|
|
if err != nil {
|
|
return
|
|
}
|
|
for _, e := range entries {
|
|
if e.Manifest.Module == module {
|
|
return
|
|
}
|
|
}
|
|
takeNewModules(ctx, open, module)
|
|
})
|
|
}
|
|
|
|
// background is the work the take-ins of keys and of announcements start beside the event that woke them, so a test
|
|
// ends only once the work its mesh started has.
|
|
var background sync.WaitGroup
|
|
|
|
// inBackground runs work beside the caller, counted in background and, under a serving controller, in its serving group,
|
|
// so serve's stop waits for it and a serve that is stopping starts none (serving_group.go). A variable so a test of what a
|
|
// walk does around a take-in may run it in the caller's turn and need not wait for it.
|
|
var inBackground = func(ctx context.Context, work func()) {
|
|
background.Add(1)
|
|
if !goServing(ctx, func() {
|
|
defer background.Done()
|
|
work()
|
|
}) {
|
|
background.Done()
|
|
}
|
|
}
|
|
|
|
// newModulesOf are the modules the indexer holds in a repository: what a first merge's word wakes.
|
|
func newModulesOf(ctx context.Context, repository string) ([]string, error) {
|
|
keys, closeKeys, err := openBuilderKeys()
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
defer closeKeys()
|
|
values, _, err := keys.Modules(ctx)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
var out []string
|
|
for module, v := range values {
|
|
if strings.EqualFold(v.Repository, repository) {
|
|
out = append(out, module)
|
|
}
|
|
}
|
|
sort.Strings(out)
|
|
return out, nil
|
|
}
|