novox/mesh-controller
18
0
Fork 0
Code Issues Pull Requests Releases Wiki Activity

Take done keys in from a durable pull consumer, a batch at a time (issue 1945, ADR 0544)

Browse Source 
The take-in watched the builder's and the indexer's buckets, started a follower
per module that retried on a 30 s timer, and at start took every catalogued
module in alphabetically: on 2026-10-25 a restart took in about 2 modules a
minute, about 190 followers queued behind 16 slots, and every take-in composed
two or three times even with nothing new, so a new build waited 30 minutes and
more for its credential.

Now two durable consumers named mesh-controller-take-in read the indexer's
modules bucket (every module's last trunk value when the consumer is made, then
each write) and the builder's builds bucket (each write), at most 16 keys
unacknowledged each. The bus keeps what is not taken in: the start's look is the
consumers' own catch-up, and a restart costs only what was left unacknowledged.
A batch reads the catalogue once, takes each module in once, and asks the
composer twice: once for its modules' nodes, once for the bus's node with the
grants. A module registered at its trunk whose compositions and grant exist, or
that this process took in at the same trunk, composes nothing; a key delivered
again is taken in whole. A wait for the builder's or indexer's next write is
acknowledged, a wait on a read or a failure refused with a 30 s delay. The
backlog is in status, and a backlog standing still for 5 minutes raises
take-in-stalled. The controller's account is granted the two consumers' acks.

The follower tests go with the followers: twelve tests become three.
Research 072: the take-ins on a bounded pull consumer (milestone 4, done keys
alone) and the start's look composing only what changed (milestone 2).
This commit is contained in:
jschoubben
2026-10-25 17:59:20 +01:00
parent 9ff47f6aaa
commit bfb271a91b
17 changed files with 926 additions and 749 deletions
Download Patch File Download Diff File Expand all files Collapse all files
cmd/mesh-controller/build.go
+3
View File
@@ -273,4 +273,7 @@ type answers struct {
// applied, is not well: an assignment somebody made is not made yet, or will not be.
pending []inventory.PendingAssignment
pendingUnread string
// takeIn is the take-in's backlog of done keys on the bus (novox/hq ADR 0544 §6), nil where it cannot be read
// (a process with no bus, or a consumer not made yet).
takeIn *takeInBacklog
}
cmd/mesh-controller/composed_integration_test.go
+4 -2
View File
@@ -102,8 +102,10 @@ func TestTheDeclarationsStreamGoesOnceEveryNodeRunsTheNewEngine(t *testing.T) {
// The new node-engine registered at its fingerprint, from its done key.
n6 := fpOf('6')
keys, trunk := aKeyFor("mesh-host", "6", n6, "novox/mesh-host", "", "f00d0123456789ab")
if _, err := liveDoneKeySteps(m.open, trunkKeys{keyBuckets: keys, trunk: trunk}).take(ctx, "mesh-host"); err != nil {
t.Fatal(err)
settled, _ := liveTakeInSteps(m.open, trunkKeys{keyBuckets: keys, trunk: trunk}, m.js).batch(ctx,
[]delivered{{module: "mesh-host", again: true}}, map[string]string{})
if len(settled) != 1 || settled[0].err != nil || settled[0].verdict != acknowledged {
t.Fatalf("the take-in of mesh-host: %+v", settled)
}
if _, err := m.js.CreateOrUpdateKeyValue(ctx, jetstream.KeyValueConfig{Bucket: sdkdelivery.HaveBucket, History: sdkdelivery.HaveHistory}); err != nil {
t.Fatal(err)
cmd/mesh-controller/composed_keys.go
+248 -306
View File
@@ -4,6 +4,8 @@ import (
"context"
"errors"
"fmt"
"maps"
"slices"
"sort"
"strings"
"sync"
@@ -12,210 +14,246 @@ import (
"git.novox.be/novox/mesh-sdk/go/catalogue"
sdkdelivery "git.novox.be/novox/mesh-sdk/go/delivery"
"github.com/nats-io/nats.go/jetstream"
"github.com/novox/mesh-controller/internal/broker"
"github.com/novox/mesh-controller/internal/builderkeys"
"github.com/novox/mesh-controller/internal/conditions"
"github.com/novox/mesh-controller/internal/inventory"
)
// A new fingerprint of a registered module, from its done key to its compositions (novox/hq ADR 0539 clauses 5 and 7,
// ADR 0501 §2). With no delivery plan left to take a module's done key in, a write of the builder's or the indexer's
// keys does it: the controller takes in the done key of the fingerprint the indexer's trunk value names (registers the
// module at it, as a plan's take-in did), composes every node the module is assigned to (which makes what a module's
// new manifest needs, a credential among it, as a send made it), composes the node holding the bus (its user list now
// carries the module's permissions at the new fingerprint: a new revision of the bus module's composition alone,
// written into its want at once by the assigner), and writes `grants.<module>.<fp>`. The assigner's sixth and seventh
// rules then let a want name the fingerprint. **No timer**: woken by the keys, and a starting controller looks at every
// module once (doneKeysAtStart).
// ADR 0501 §2, ADR 0544). With no delivery plan left to take a module's done key in, a write of the builder's or the
// indexer's keys does it: the controller takes in the done key of the fingerprint the indexer's trunk value names
// (registers the module at it, as a plan's take-in did), composes every node the module is assigned to (which makes what
// a module's new manifest needs, a credential among it), composes the node holding the bus (its user list now carries
// the module's permissions at the new fingerprint), and writes `grants.<module>.<fp>`. The assigner's sixth and seventh
// rules then let a want name the fingerprint.
//
// **A durable pull consumer, not a watch** (ADR 0544; research 072, milestones 2 and 4). The writes are read from two
// durable consumers (take_in_consumer.go): the bus keeps what is not taken in yet, so a restart takes in only what the
// controller before it left unacknowledged, and the start's look is the consumers' own catch-up. A batch of keys asks the
// composer twice whatever its size, and a key with nothing new composes nothing.
// doneKeySteps are what taking a module's done key in does, each a step a test may stand in for.
type doneKeySteps struct {
// held says the catalogue holds the module (a new one is the new-module take-in's, new_modules.go).
held func(ctx context.Context, module string) (bool, error)
// takeInSteps are what taking a batch of done keys in does, each a step a test may stand in for.
type takeInSteps struct {
// catalogue is every module the catalogue holds, with the nodes it is assigned to: read once a batch. A module it
// does not hold is the new-module take-in's (new_modules.go).
catalogue func(ctx context.Context) (map[string][]string, error)
// trunk is the indexer's trunk value of the module.
trunk func(ctx context.Context, module string) (sdkdelivery.Module, error)
// registeredAt says the module is registered at the commit.
registeredAt func(ctx context.Context, module, commit string) (bool, error)
// takeIn registers the module from the builder's done key of the trunk's fingerprint (takeInKey); errKeyWaits when
// the key is not done yet.
// takeIn registers the module from the builder's done key of the trunk's fingerprint (takeInKey).
takeIn func(ctx context.Context, module string, trunk sdkdelivery.Module) error
// nodesOf are the nodes the module is assigned to; busHolder the node holding the bus, "" for none.
nodesOf func(ctx context.Context, module string) ([]string, error)
// taken says the module's compositions at fp exist for every node and its grant is written: nothing to compose.
taken func(ctx context.Context, module, fp string, nodes []string) (bool, error)
// busHolder is the node holding the bus, "" for none.
busHolder func(ctx context.Context) (string, error)
// compose writes the nodes' compositions; grants writes the grants issued.
compose func(ctx context.Context, nodes []string) error
grants func(ctx context.Context) error
// compose composes the nodes, and writes the grants after when grants: each node's error, and the grants'.
compose func(ctx context.Context, nodes []string, grants bool) (map[string]error, error, error)
}
// take takes module's done key in and composes what it changed. It answers what it did in a line, "" when there was
// nothing to do.
func (s doneKeySteps) take(ctx context.Context, module string) (string, error) {
held, err := s.held(ctx, module)
if err != nil || !held {
return "", err
}
trunk, err := s.trunk(ctx, module)
if errors.Is(err, builderkeys.ErrNoValue) {
return "", nil
}
if err != nil {
return "", err
}
if !strings.HasPrefix(trunk.FP2, sdkdelivery.SourcePrefix2) {
return "", nil
}
at, err := s.registeredAt(ctx, module, trunk.Commit)
if err != nil {
return "", err
}
said := ""
if !at {
if err := s.takeIn(ctx, module, trunk); err != nil {
return "", err
}
said = fmt.Sprintf("%s registered from the builder's done key %s (%s)", module, short(trunk.FP2), short(trunk.Commit))
}
nodes, err := s.nodesOf(ctx, module)
if err != nil {
return said, err
}
if len(nodes) == 0 {
return said, nil
}
// The module's nodes first: composing them makes what the new manifest needs, its credentials among it.
if err := s.compose(ctx, nodes); err != nil {
return said, err
}
// Then the bus's user list, which carries the module's permissions at its new fingerprint.
holder, err := s.busHolder(ctx)
if err != nil {
return said, err
}
if holder != "" && !inList(nodes, holder) {
if err := s.compose(ctx, []string{holder}); err != nil {
return said, err
}
}
// Then the grant, which with the composition lets a want name the fingerprint (rules 6 and 7).
if err := s.grants(ctx); err != nil {
return said, err
}
if said == "" {
said = fmt.Sprintf("%s at %s", module, short(trunk.FP2))
}
return said + "; composed for " + strings.Join(nodes, ", ") + ", its grant written", nil
// delivered is one key a consumer delivered: its module, and whether it was delivered before (a take-in that may have
// stopped half-way, so it is taken in whole).
type delivered struct {
module string
again bool
}
func inList(list []string, s string) bool {
for _, v := range list {
if v == s {
return true
}
}
return false
}
// verdict is what becomes of a module's delivered keys.
type verdict int
// doneKeyTakeOnce is one take-in; a variable so a test may stand in for it. doneKeysSay says a line of the take-ins.
var (
doneKeyTakeOnce = takeDoneKeyOnce
doneKeysSay = func(s string) { fmt.Println(s) }
const (
// acknowledged: taken in, nothing new, a wait a later write ends, or a build key that failed.
acknowledged verdict = iota
// refused: delivered again by the bus after doneKeyRetry (a wait on a read, or a failure).
refused
)
// doneKeysBeforeLetGo is called just before a follower lets go of its module; a variable so a test can wake the module
// at that moment. Nil does nothing.
var doneKeysBeforeLetGo func(module string)
// doneKeysLookAgain marks a module woken while a take-in of it ran: that take-in goes round once more.
var doneKeysLookAgain sync.Map
// doneKeysMu keeps one take-in of a module's done key at a time in this process, per module.
var doneKeysMu sync.Map
// takeDoneKey is a woken module's take-in on the live stores and buckets, said. A take-in that fails (its compositions
// or its grant not written) is tried again every doneKeyRetry while it fails, loudly: printed each time, and an urgent
// condition raised by the first until it has worked, as the chain's own moves are (layBuildsOutSoon). A key not done
// yet is no failure: it is said once for each reason it waits for and looked at again every doneKeyRetry.
func takeDoneKey(ctx context.Context, open *stores, module string) {
mu, _ := doneKeysMu.LoadOrStore(module, &sync.Mutex{})
for {
if !mu.(*sync.Mutex).TryLock() {
// A take-in of the module already follows it: it is told to look again once it is done, so a newer key
// written meanwhile is never lost (the review of issue 1945).
doneKeysLookAgain.Store(module, true)
return
}
followDoneKey(ctx, open, module)
if doneKeysBeforeLetGo != nil {
doneKeysBeforeLetGo(module)
}
mu.(*sync.Mutex).Unlock()
// A wake between this follower's last look and its letting go failed its TryLock and only marked the module:
// looked at again now that it is let go, and taken once more (the review of issue 1945).
if _, again := doneKeysLookAgain.LoadAndDelete(module); !again || ctx.Err() != nil {
return
}
}
// settledKey is what a batch did with one module.
type settledKey struct {
module string
verdict verdict
// taken: registered, composed and granted now (not nothing new); its condition is cleared.
taken bool
// nothingNew: registered and composed before, so nothing was done.
nothingNew bool
// waits is why it waits, "" when it does not.
waits string
// err is a failure: raised as the module's condition.
err error
// said is the line said of it, "" for none.
said string
}
// followDoneKey takes a module's done key in, under the module's lock, until it is taken in or ctx ends: a failure or a
// wait is tried again every doneKeyRetry.
func followDoneKey(ctx context.Context, open *stores, module string) {
failed := false
saidWait := ""
for {
doneKeysLookAgain.Delete(module) // what this look reads covers every wake before it
err := takeOnceInSlot(ctx, open, module)
o := doneKeyObservation(module)
if err == nil {
// Cleared on every success, whoever raised it: a controller before a restart or another follower (issue
// 1945). A condition that is not open costs one read.
if keeper := conditionsFrom; keeper != nil {
_, _ = keeper.Clear(ctx, o.Key(), "composed and granted")
}
failed = false
if _, again := doneKeysLookAgain.LoadAndDelete(module); again {
continue // a wake came while this one ran: once more, for the key written meanwhile
}
return
// batch takes the delivered keys in, each module once: it answers one settledKey per module in the order first
// delivered, and the number of composer passes it asked. memo is, per module, the trunk (fingerprint and commit) this
// process took in last; a module at the same trunk delivered for the first time has nothing new.
func (s takeInSteps) batch(ctx context.Context, keys []delivered, memo map[string]string) ([]settledKey, int) {
var order []string
again := map[string]bool{}
for _, k := range keys {
if _, seen := again[k.module]; !seen {
order = append(order, k.module)
}
if errors.Is(err, errKeyWaits) {
// A take-in that waits is said once for each reason, and tried again every doneKeyRetry: the key write that
// would wake it may never come (issue 1945, live on 2026-10-25: mesh-host's take-in waited unsaid).
if why := err.Error(); why != saidWait {
doneKeysSay(fmt.Sprintf("done keys: %s waits: %s; looked at again every %s", module,
strings.TrimPrefix(why, errKeyWaits.Error()+": "), doneKeyRetry))
saidWait = why
}
} else {
doneKeysSay(fmt.Sprintf("done keys: %s: %v; tried again in %s", module, err, doneKeyRetry))
again[k.module] = again[k.module] || k.again
}
out := make([]settledKey, 0, len(order))
all := func(v verdict, err error) ([]settledKey, int) {
for _, m := range order {
out = append(out, settledKey{module: m, verdict: v, err: err})
}
if errors.Is(err, errKeyWaits) {
select {
case <-ctx.Done():
return
case <-time.After(doneKeyRetry):
}
return out, 0
}
held, err := s.catalogue(ctx)
if err != nil {
return all(refused, fmt.Errorf("the catalogue cannot be read: %w", err))
}
type toCompose struct {
at int
nodes []string
mark string
said string
}
var need []toCompose
for _, m := range order {
r := settledKey{module: m}
nodes, isHeld := held[m]
if !isHeld {
out = append(out, r)
continue
}
if ctx.Err() != nil {
return // stopped: a take-in cut by the stop is no fault of the build, and the next controller takes it in
trunk, err := s.trunk(ctx, m)
if errors.Is(err, builderkeys.ErrNoValue) || (err == nil && !strings.HasPrefix(trunk.FP2, sdkdelivery.SourcePrefix2)) {
out = append(out, r)
continue
}
if keeper := conditionsFrom; keeper != nil && !failed && !errors.Is(err, context.Canceled) {
// A take-in whose own step was cancelled is tried again, and raises nothing: a cancellation is no fault
// of the build (issue 1945: 30 urgent conditions live, each a take-in cut by a restart).
o.Summary += err.Error()
if _, oerr := keeper.Observe(ctx, o); oerr == nil {
failed = true
if err != nil {
r.verdict, r.err = refused, err
out = append(out, r)
continue
}
mark := trunk.FP2 + "@" + trunk.Commit
if !again[m] && memo[m] == mark {
r.nothingNew = true
out = append(out, r)
continue
}
at, err := s.registeredAt(ctx, m, trunk.Commit)
if err != nil {
r.verdict, r.err = refused, err
out = append(out, r)
continue
}
said := ""
if !at {
if err := s.takeIn(ctx, m, trunk); err != nil {
switch {
case errors.Is(err, errWaitsForAWrite):
r.waits = err.Error()
case errors.Is(err, errKeyWaits):
r.verdict, r.waits = refused, err.Error()
case errors.Is(err, errKeyFailed):
r.err = err // acknowledged: only a new build ends it, and its key is delivered then
default:
r.verdict, r.err = refused, err
}
out = append(out, r)
continue
}
said = fmt.Sprintf("%s registered from the builder's done key %s (%s)", m, short(trunk.FP2), short(trunk.Commit))
} else if !again[m] {
// Registered at its trunk already: composed and granted by a take-in before this one, or at this
// controller's start, unless that take-in stopped half-way, which leaves a composition or the grant missing.
if done, err := s.taken(ctx, m, trunk.FP2, nodes); err == nil && done {
memo[m] = mark
r.nothingNew = true
out = append(out, r)
continue
}
}
select {
case <-ctx.Done():
return
case <-time.After(doneKeyRetry):
if len(nodes) == 0 {
memo[m] = mark
r.taken, r.said = true, said
out = append(out, r)
continue
}
if said == "" {
said = fmt.Sprintf("%s at %s", m, short(trunk.FP2))
}
need = append(need, toCompose{at: len(out), nodes: nodes, mark: mark, said: said})
out = append(out, r)
}
if len(need) == 0 {
return out, 0
}
refuseNeed := func(err error) ([]settledKey, int) {
for _, n := range need {
out[n.at].verdict, out[n.at].err = refused, err
}
return out, 0
}
holder, err := s.busHolder(ctx)
if err != nil {
return refuseNeed(fmt.Errorf("the node holding the bus cannot be read: %w", err))
}
// The modules' nodes first, in one pass: composing them makes what their new manifests need, credentials among it.
var union []string
for _, n := range need {
for _, node := range n.nodes {
if node != holder && !slices.Contains(union, node) {
union = append(union, node)
}
}
}
sort.Strings(union)
passes := 0
failed := map[string]error{}
if len(union) > 0 {
f, _, err := s.compose(ctx, union, false)
passes++
if err != nil {
return refuseNeed(err)
}
maps.Copy(failed, f)
}
// Then the bus's user list, which carries the modules' permissions at their new fingerprints, and the grants, which
// with the compositions let a want name each fingerprint (rules 6 and 7).
var bus []string
if holder != "" {
bus = []string{holder}
}
f, grantsErr, err := s.compose(ctx, bus, true)
passes++
if err != nil {
return refuseNeed(err)
}
maps.Copy(failed, f)
for _, n := range need {
var errs []error
for _, node := range append(slices.Clone(n.nodes), bus...) {
if e := failed[node]; e != nil && !slices.ContainsFunc(errs, func(x error) bool { return x == e }) {
errs = append(errs, e)
}
}
if grantsErr != nil {
errs = append(errs, grantsErr)
}
r := &out[n.at]
if len(errs) > 0 {
r.verdict, r.err = refused, errors.Join(errs...)
continue
}
memo[r.module] = n.mark
r.taken = true
r.said = n.said + "; composed for " + strings.Join(n.nodes, ", ") + ", its grant written"
}
return out, passes
}
// doneKeyRetry is how long a failed take-in of a done key waits before it tries again.
// doneKeysSay says a line of the take-ins; a variable so a test may hear it.
var doneKeysSay = func(s string) { fmt.Println(s) }
// doneKeyRetry is how long the bus waits before it delivers again a key whose take-in was refused.
var doneKeyRetry = 30 * time.Second
// kindNotComposed is the condition of a module's new fingerprint whose compositions or grant are not written: until
@@ -228,30 +266,20 @@ func doneKeyObservation(module string) conditions.Observation {
Summary: module + "'s new fingerprint is not composed or granted, so no want can name it: "}
}
// takeDoneKeyOnce is one take-in of a module's done key.
func takeDoneKeyOnce(ctx context.Context, open *stores, module string) error {
keys, closeKeys, err := openBuilderKeys()
if err != nil {
return fmt.Errorf("the builder's keys cannot be read: %w", err)
}
defer closeKeys()
said, err := liveDoneKeySteps(open, keys).take(ctx, module)
if err == nil && said != "" {
fmt.Printf("done keys: %s\n", said)
}
return err
}
// liveDoneKeySteps are the steps on the controller's own stores and the bus.
func liveDoneKeySteps(open *stores, keys registerKeys) doneKeySteps {
// liveTakeInSteps are the steps on the controller's own stores and the bus.
func liveTakeInSteps(open *stores, keys registerKeys, js jetstream.JetStream) takeInSteps {
inv := open.inventory
return doneKeySteps{
held: func(ctx context.Context, module string) (bool, error) {
_, err := inv.SourceOf(ctx, module)
if errors.Is(err, inventory.ErrNoSuchModule) {
return false, nil
return takeInSteps{
catalogue: func(ctx context.Context) (map[string][]string, error) {
entries, err := inv.Catalogued(ctx)
if err != nil {
return nil, err
}
return err == nil, err
out := make(map[string][]string, len(entries))
for _, e := range entries {
out[e.Manifest.Module] = e.On
}
return out, nil
},
trunk: func(ctx context.Context, module string) (sdkdelivery.Module, error) {
return keys.ModuleAt(ctx, module, "")
@@ -270,17 +298,8 @@ func liveDoneKeySteps(open *stores, keys registerKeys) doneKeySteps {
}
return err
},
nodesOf: func(ctx context.Context, module string) ([]string, error) {
entries, err := inv.Catalogued(ctx)
if err != nil {
return nil, err
}
for _, e := range entries {
if e.Manifest.Module == module {
return e.On, nil
}
}
return nil, nil
taken: func(ctx context.Context, module, fp string, nodes []string) (bool, error) {
return composedAndGranted(ctx, js, module, fp, nodes)
},
busHolder: func(ctx context.Context) (string, error) {
holders, err := seatHolders(ctx, inv)
@@ -291,123 +310,46 @@ func liveDoneKeySteps(open *stores, keys registerKeys) doneKeySteps {
},
// The composer's, never the take-in's own: one pass at a time composes each node asked for once, and writes the
// grants once (issue 1945, composer.go).
compose: func(ctx context.Context, nodes []string) error {
return composerFor(open).compose(ctx, nodes, false)
},
grants: func(ctx context.Context) error {
return composerFor(open).compose(ctx, nil, true)
compose: func(ctx context.Context, nodes []string, grants bool) (map[string]error, error, error) {
return composerFor(open).composeEach(ctx, nodes, grants)
},
}
}
// watchDoneKeys wakes the take-in of a module's done key for each module the builder or the indexer writes, until ctx
// ends: never on the watch's own goroutine.
func watchDoneKeys(ctx context.Context, open *stores) {
keys, closeKeys, err := openBuilderKeys()
// composedAndGranted says the bus holds module's grant at fp and its composition at fp for every node: a few reads of
// keys, never a composition.
func composedAndGranted(ctx context.Context, js jetstream.JetStream, module, fp string, nodes []string) (bool, error) {
grants, err := js.KeyValue(ctx, broker.GrantsBucket)
if err != nil {
fmt.Printf("done keys: the builder's keys are not watched: %v\n", err)
return
return false, err
}
if err := builderkeys.Watch(ctx, keys.JS, func(module string) {
inBackground(ctx, func() { takeDoneKey(ctx, open, module) })
}); err != nil {
closeKeys()
fmt.Printf("done keys: the builder's keys are not watched: %v\n", err)
return
}
goServing(ctx, func() {
<-ctx.Done()
closeKeys()
})
// And every module once now, after the watch is open so no write falls between the two.
inBackground(ctx, func() {
doneKeysAtStart(ctx, catalogedModules(open), func(module string) { takeDoneKeyAtStart(ctx, open, module) })
})
}
// doneKeysAtStart takes in the done key of every catalogued module once, one after another: the start's look the watch of
// new writes cannot give (it hears updates only). A done key written while no controller served, or whose take-in the
// controller this one replaced was stopped in, is taken here; without it the module stays in the node's own composition
// with no credential until its keys are written again (issue 1945, live on 2026-10-25 at 02:34). A take-in with nothing
// to do does nothing; a catalogue that cannot be read is said.
func doneKeysAtStart(ctx context.Context, modules func(context.Context) ([]string, error), take func(module string)) {
names, err := modules(ctx)
key, err := sdkdelivery.ControllerGrantKey(module, fp)
if err != nil {
fmt.Printf("done keys: the modules are not looked at once at the start: %v\n", err)
return
return false, err
}
sort.Strings(names)
// Side by side, a few at a time: their compositions meet in the composer's passes, each node composed once per pass
// (issue 1945), so the start's look is not one module's composition after another.
sem := make(chan struct{}, startTakeIns)
var wg sync.WaitGroup
for _, m := range names {
if ctx.Err() != nil {
break
if _, err := grants.Get(ctx, key); err != nil {
if errors.Is(err, jetstream.ErrKeyNotFound) {
return false, nil
}
sem <- struct{}{}
wg.Add(1)
go func() {
defer func() { <-sem; wg.Done() }()
take(m)
}()
return false, err
}
wg.Wait()
}
// startTakeIns is how many take-ins the start's look runs at once, and how many run at once in all.
const startTakeIns = 16
// takeInSlots are the take-ins that may run at once in this process, the watch's and the start's look's together: a
// burst of done keys (about 120 at 16:07:36 on 2026-10-25) once started a take-in each, all at once, and their reads
// filled the store until it refused every connection (issues 1945 and 1277).
var takeInSlots = make(chan struct{}, startTakeIns)
// takeOnceInSlot is one take-in once a slot is free; a take-in that waits for its next try holds none.
func takeOnceInSlot(ctx context.Context, open *stores, module string) error {
select {
case takeInSlots <- struct{}{}:
case <-ctx.Done():
return ctx.Err()
digests, err := js.KeyValue(ctx, sdkdelivery.ControllerComposedBucket)
if err != nil {
return false, err
}
defer func() { <-takeInSlots }()
return doneKeyTakeOnce(ctx, open, module)
}
// takeDoneKeyAtStart is one take-in of the start's look, under the module's lock; one that fails is handed to takeDoneKey,
// which tries again every doneKeyRetry, loudly, so a failing module never holds up the look at the others.
func takeDoneKeyAtStart(ctx context.Context, open *stores, module string) {
mu, _ := doneKeysMu.LoadOrStore(module, &sync.Mutex{})
if !mu.(*sync.Mutex).TryLock() {
// A take-in already follows the module: told to look again, it takes what the start would have (issue 1945).
doneKeysLookAgain.Store(module, true)
return
}
doneKeysLookAgain.Delete(module) // what this look reads covers every wake before it
err := takeOnceInSlot(ctx, open, module)
mu.(*sync.Mutex).Unlock()
// Read after letting go, as every holder of a module's lock does: a wake during this look only marked the module
// (the review of issue 1945).
_, again := doneKeysLookAgain.LoadAndDelete(module)
if err != nil || again {
// A failure, a wait or a wake meanwhile is followed by takeDoneKey, which says it and looks again.
inBackground(ctx, func() { takeDoneKey(ctx, open, module) })
}
}
// catalogedModules are the names of every module the catalogue holds.
func catalogedModules(open *stores) func(context.Context) ([]string, error) {
return func(ctx context.Context) ([]string, error) {
entries, err := open.inventory.Catalogued(ctx)
for _, node := range nodes {
key, err := sdkdelivery.ComposedKey(node, module, fp)
if err != nil {
return nil, err
return false, err
}
out := make([]string, 0, len(entries))
for _, e := range entries {
out = append(out, e.Manifest.Module)
if _, err := digests.Get(ctx, key); err != nil {
if errors.Is(err, jetstream.ErrKeyNotFound) {
return false, nil
}
return false, err
}
return out, nil
}
return true, nil
}
// doneKeysComposers are the composers of the done keys' take-ins in this process, one for each set of stores (a test
cmd/mesh-controller/composed_keys_cancel_test.go
-83
View File
@@ -1,83 +0,0 @@
package main
import (
"context"
"fmt"
"testing"
"time"
"github.com/novox/mesh-controller/internal/conditions"
)
// A take-in cut by a cancellation raises no condition (issue 1945: on 2026-10-25 thirty urgent fingerprint-not-composed
// conditions were live, each a take-in a stop or a restart cut with "context canceled"): when the controller stops, the
// next one takes it in; when a step of its own was cancelled, it is tried again.
func TestATakeInCutByACancellationRaisesNothing(t *testing.T) {
for _, tc := range []struct {
name string
stop bool // the controller's context ends, else only the step's error is a cancellation
}{{"the controller stops", true}, {"a step was cancelled", false}} {
t.Run(tc.name, func(t *testing.T) {
was, wasRetry, wasSay, wasKeeper := doneKeyTakeOnce, doneKeyRetry, doneKeysSay, conditionsFrom
t.Cleanup(func() {
doneKeyTakeOnce, doneKeyRetry, doneKeysSay, conditionsFrom = was, wasRetry, wasSay, wasKeeper
})
store := conditions.NewInMemory()
k := conditions.NewKeeper(t.Context(), conditions.Options{Store: store, History: store,
Teller: &conditions.Told{}})
t.Cleanup(func() { k.Close(context.Background()) })
conditionsFrom = k
doneKeyRetry = time.Millisecond
doneKeysSay = func(string) {}
ctx, cancel := context.WithCancel(context.Background())
defer cancel()
calls := 0
doneKeyTakeOnce = func(context.Context, *stores, string) error {
calls++
if tc.stop {
cancel() // stopped mid-way
return fmt.Errorf("the module cannot be composed: %w", context.Canceled)
}
if calls == 1 {
return fmt.Errorf("the module cannot be composed: %w", context.Canceled)
}
return nil
}
done := make(chan struct{})
go func() { takeDoneKey(ctx, nil, "cut-module"); close(done) }()
select {
case <-done:
case <-time.After(5 * time.Second):
t.Fatal("the take-in did not end")
}
if !tc.stop && calls != 2 {
t.Errorf("a take-in whose step was cancelled was tried %d time(s), want 2", calls)
}
if c, found, err := conditions.ReadOne(t.Context(), store, doneKeyObservation("cut-module").Key()); err != nil || found {
t.Fatalf("a take-in cut by a cancellation raised %+v (found %v, %v)", c, found, err)
}
})
}
}
// A take-in that succeeds clears the module's fingerprint-not-composed condition whoever raised it: a controller before a
// restart, or another follower (issue 1945: the operator read "a build needs a look" for take-ins that had since worked).
func TestASucceedingTakeInClearsAConditionAnotherRaised(t *testing.T) {
was, wasRetry, wasSay, wasKeeper := doneKeyTakeOnce, doneKeyRetry, doneKeysSay, conditionsFrom
t.Cleanup(func() { doneKeyTakeOnce, doneKeyRetry, doneKeysSay, conditionsFrom = was, wasRetry, wasSay, wasKeeper })
store := conditions.NewInMemory()
k := conditions.NewKeeper(t.Context(), conditions.Options{Store: store, History: store, Teller: &conditions.Told{}})
t.Cleanup(func() { k.Close(context.Background()) })
conditionsFrom = k
doneKeysSay = func(string) {}
o := doneKeyObservation("raised-before")
o.Summary += "the store refused the connection"
if _, err := k.Observe(t.Context(), o); err != nil {
t.Fatal(err)
}
doneKeyTakeOnce = func(context.Context, *stores, string) error { return nil }
takeDoneKey(t.Context(), nil, "raised-before")
if c, found, err := conditions.ReadOne(t.Context(), store, o.Key()); err != nil || found {
t.Fatalf("a take-in that succeeded left the condition another raised: %+v (found %v, %v)", c, found, err)
}
}
cmd/mesh-controller/composed_keys_cap_test.go
-58
View File
@@ -1,58 +0,0 @@
package main
import (
"context"
"fmt"
"sync"
"sync/atomic"
"testing"
"time"
)
// A burst of done keys runs at most startTakeIns take-ins at once, the watch's and the start's look together: at
// 16:07:36 on 2026-10-25 the builder rewrote about 120 done keys at once, each woke its own take-in, and each opened
// pools of its own until novox's store refused every connection (issues 1945 and 1277).
func TestABurstOfDoneKeysRunsAtMostSixteenTakeInsAtOnce(t *testing.T) {
was := doneKeyTakeOnce
t.Cleanup(func() { doneKeyTakeOnce = was })
var running, most atomic.Int64
var taken atomic.Int64
doneKeyTakeOnce = func(context.Context, *stores, string) error {
now := running.Add(1)
for {
m := most.Load()
if now <= m || most.CompareAndSwap(m, now) {
break
}
}
time.Sleep(2 * time.Millisecond)
running.Add(-1)
taken.Add(1)
return nil
}
ctx := context.Background()
var wg sync.WaitGroup
// The watch's wakes: one take-in each, in the background, as watchDoneKeys starts them.
for i := range 120 {
wg.Add(1)
go func() { defer wg.Done(); takeDoneKey(ctx, nil, fmt.Sprintf("burst-%03d", i)) }()
}
// And the start's look at the same moment, over other modules.
wg.Add(1)
go func() {
defer wg.Done()
var start []string
for i := range 40 {
start = append(start, fmt.Sprintf("start-%03d", i))
}
doneKeysAtStart(ctx, func(context.Context) ([]string, error) { return start, nil },
func(m string) { takeDoneKeyAtStart(ctx, nil, m) })
}()
wg.Wait()
if got := taken.Load(); got != 160 {
t.Fatalf("took in %d modules, want 160", got)
}
if m := most.Load(); m > startTakeIns {
t.Fatalf("ran %d take-ins at once, want at most %d", m, startTakeIns)
}
}
cmd/mesh-controller/composed_keys_start_test.go
-34
View File
@@ -1,34 +0,0 @@
package main
import (
"context"
"errors"
"reflect"
"sort"
"sync"
"testing"
)
// A starting controller takes in the done key of every catalogued module once, beside the watch of new writes: a done key
// written while no controller served, or whose take-in the controller it replaced was stopped in, is otherwise never
// taken, and its module stays in the node's own composition with no credential (issue 1945, live on 2026-10-25 at 02:34:
// the controller replaced mid take-in left mesh-provisioner-client, gitea, postgres and many more unregistered at their
// new fingerprints, and the new one, watching updates only, never looked).
func TestAStartingControllerTakesEveryModulesDoneKeyOnce(t *testing.T) {
var took []string
var mu sync.Mutex
doneKeysAtStart(context.Background(), func(context.Context) ([]string, error) {
return []string{"mesh-provisioner-client", "gitea", "postgres"}, nil
}, func(module string) { mu.Lock(); took = append(took, module); mu.Unlock() })
sort.Strings(took)
if want := []string{"gitea", "mesh-provisioner-client", "postgres"}; !reflect.DeepEqual(took, want) {
t.Fatalf("took %v, want %v", took, want)
}
// A catalogue that cannot be read takes nothing and is said, never fatal.
took = nil
doneKeysAtStart(context.Background(), func(context.Context) ([]string, error) { return nil, errors.New("down") },
func(module string) { took = append(took, module) })
if len(took) != 0 {
t.Fatalf("took %v from a catalogue that could not be read", took)
}
}
cmd/mesh-controller/composed_keys_test.go
+217 -68
View File
@@ -4,84 +4,245 @@ import (
"context"
"encoding/json"
"errors"
"fmt"
"slices"
"strings"
"testing"
"time"
sdkdelivery "git.novox.be/novox/mesh-sdk/go/delivery"
"github.com/novox/mesh-controller/internal/builderkeys"
)
// stepsFor are doneKeySteps over a module registered at registered, whose trunk value names fp at commit, assigned to
// nodes, the bus held on holder; every act is written to acts in order.
func stepsFor(acts *[]string, registered, commit, fp string, nodes []string, holder string) doneKeySteps {
return doneKeySteps{
held: func(_ context.Context, module string) (bool, error) { return module == "mesh-mail", nil },
trunk: func(context.Context, string) (sdkdelivery.Module, error) {
if fp == "" {
// aTakeIn is a mesh for the take-in's steps: per module its registered commit, its trunk (commit, fingerprint), its
// nodes, whether its compositions and grant exist, and what its take-in answers; every act is written to acts.
type aTakeIn struct {
registered map[string]string
trunks map[string]string // module -> commit; no entry: no trunk value
nodes map[string][]string
taken map[string]bool
takeInErr map[string]error
composeErr map[string]error
grantsErr error
holder string
acts []string
}
func (m *aTakeIn) steps() takeInSteps {
fp := func(module string) string { return sdkdelivery.SourcePrefix2 + strings.Repeat(module[:1], 64) }
return takeInSteps{
catalogue: func(context.Context) (map[string][]string, error) {
m.acts = append(m.acts, "catalogue")
out := map[string][]string{}
for k, v := range m.nodes {
out[k] = v
}
return out, nil
},
trunk: func(_ context.Context, module string) (sdkdelivery.Module, error) {
commit, ok := m.trunks[module]
if !ok {
return sdkdelivery.Module{}, builderkeys.ErrNoValue
}
return sdkdelivery.Module{Commit: commit, FP2: fp}, nil
return sdkdelivery.Module{Commit: commit, FP2: fp(module)}, nil
},
registeredAt: func(_ context.Context, module, commit string) (bool, error) {
return m.registered[module] == commit, nil
},
registeredAt: func(_ context.Context, _ string, c string) (bool, error) { return c == registered, nil },
takeIn: func(_ context.Context, module string, trunk sdkdelivery.Module) error {
*acts = append(*acts, "take in "+module+" at "+trunk.Commit)
return nil
m.acts = append(m.acts, "take in "+module+" at "+trunk.Commit)
return m.takeInErr[module]
},
nodesOf: func(context.Context, string) ([]string, error) { return nodes, nil },
busHolder: func(context.Context) (string, error) { return holder, nil },
compose: func(_ context.Context, ns []string) error {
*acts = append(*acts, "compose "+strings.Join(ns, ","))
return nil
taken: func(_ context.Context, module, _ string, _ []string) (bool, error) { return m.taken[module], nil },
busHolder: func(context.Context) (string, error) { return m.holder, nil },
compose: func(_ context.Context, nodes []string, grants bool) (map[string]error, error, error) {
act := "compose " + strings.Join(nodes, ",")
if grants {
act += " and grants"
}
m.acts = append(m.acts, act)
failed := map[string]error{}
for _, n := range nodes {
if e := m.composeErr[n]; e != nil {
failed[n] = e
}
}
if grants {
return failed, m.grantsErr, nil
}
return failed, nil, nil
},
grants: func(context.Context) error { *acts = append(*acts, "grants"); return nil },
}
}
// 1945/AC3 (unit half): a done key of a new fingerprint of an assigned module is taken in, then the module's nodes are
// composed, then the node holding the bus (its user list), then the grants: the grant and the composition exist
// before the assigner's rules let a want name the fingerprint. A module registered at the trunk's commit already is
// composed and granted without a second take-in; one assigned nowhere, unregistered, or with no trunk value is left
// alone.
func TestADoneKeyIsTakenInComposedAndGranted(t *testing.T) {
fp := sdkdelivery.SourcePrefix2 + strings.Repeat("c", 64)
var acts []string
said, err := stepsFor(&acts, "old", "new", fp, []string{"ace", "g14"}, "novox").take(context.Background(), "mesh-mail")
if err != nil {
t.Fatal(err)
// A batch of done keys takes each module in once and asks the composer at most twice, whatever its size: once for the
// nodes of every module with something new, then once for the node holding the bus with the grants (ADR 0544 §5). A
// module registered at its trunk whose compositions and grant exist, or that this process took in at the same trunk,
// composes nothing (§3): the old take-in composed it three times over, every key and every start (issue 1945, 17:36 on
// 2026-10-25). A key delivered again is taken in whole. A wait for a write is acknowledged, a wait on a read and a
// failure refused, a failed build key acknowledged with its condition (§4).
func TestATakeInBatch(t *testing.T) {
waitsForAWrite := fmt.Errorf("%w: the key is building", errWaitsForAWrite)
waitsOnARead := fmt.Errorf("%w: the store is away", errKeyWaits)
keyFailed := fmt.Errorf("%w: the build failed", errKeyFailed)
type want struct {
verdict verdict
taken, nothingNew bool
waits, err bool
}
if got := strings.Join(acts, "; "); got != "take in mesh-mail at new; compose ace,g14; compose novox; grants" {
t.Errorf("the acts were %q", got)
cases := []struct {
name string
mesh aTakeIn
keys []delivered
memo map[string]string
acts []string
passes int
want map[string]want
}{
{name: "a new fingerprint: taken in, its nodes, then the bus's node and the grants",
mesh: aTakeIn{registered: map[string]string{"mail": "old"}, trunks: map[string]string{"mail": "new"},
nodes: map[string][]string{"mail": {"ace", "g14", "novox"}}, holder: "novox"},
keys: []delivered{{module: "mail"}},
acts: []string{"catalogue", "take in mail at new", "compose ace,g14", "compose novox and grants"}, passes: 2,
want: map[string]want{"mail": {taken: true}}},
{name: "registered, composed and granted: nothing composed (fails on the old take-in)",
mesh: aTakeIn{registered: map[string]string{"mail": "new", "nats": "n"}, trunks: map[string]string{"mail": "new", "nats": "n"},
nodes: map[string][]string{"mail": {"ace"}, "nats": {"novox"}}, taken: map[string]bool{"mail": true, "nats": true},
holder: "novox"},
keys: []delivered{{module: "mail"}, {module: "nats"}, {module: "mail"}},
acts: []string{"catalogue"},
want: map[string]want{"mail": {nothingNew: true}, "nats": {nothingNew: true}}},
{name: "taken in by this process at the same trunk: nothing new, not even a look at its keys",
mesh: aTakeIn{registered: map[string]string{"mail": "new"}, trunks: map[string]string{"mail": "new"},
nodes: map[string][]string{"mail": {"ace"}}, holder: "novox"},
keys: []delivered{{module: "mail"}},
memo: map[string]string{"mail": sdkdelivery.SourcePrefix2 + strings.Repeat("m", 64) + "@new"},
acts: []string{"catalogue"}, want: map[string]want{"mail": {nothingNew: true}}},
{name: "delivered again: taken in whole, even when it looks done",
mesh: aTakeIn{registered: map[string]string{"mail": "new"}, trunks: map[string]string{"mail": "new"},
nodes: map[string][]string{"mail": {"ace"}}, taken: map[string]bool{"mail": true}, holder: "novox"},
keys: []delivered{{module: "mail", again: true}},
acts: []string{"catalogue", "compose ace", "compose novox and grants"}, passes: 2,
want: map[string]want{"mail": {taken: true}}},
{name: "a take-in that stopped half-way: composed",
mesh: aTakeIn{registered: map[string]string{"mail": "new"}, trunks: map[string]string{"mail": "new"},
nodes: map[string][]string{"mail": {"ace"}}, holder: "novox"},
keys: []delivered{{module: "mail"}},
acts: []string{"catalogue", "compose ace", "compose novox and grants"}, passes: 2,
want: map[string]want{"mail": {taken: true}}},
{name: "three modules, one pass for their nodes and one for the bus",
mesh: aTakeIn{registered: map[string]string{}, trunks: map[string]string{"mail": "1", "web": "2", "bus": "3"},
nodes: map[string][]string{"mail": {"ace", "novox"}, "web": {"g14", "ace"}, "bus": {"novox"}}, holder: "novox"},
keys: []delivered{{module: "mail"}, {module: "web"}, {module: "bus"}},
acts: []string{"catalogue", "take in mail at 1", "take in web at 2", "take in bus at 3", "compose ace,g14",
"compose novox and grants"}, passes: 2,
want: map[string]want{"mail": {taken: true}, "web": {taken: true}, "bus": {taken: true}}},
{name: "assigned nowhere, held by no catalogue, no trunk value: nothing composed",
mesh: aTakeIn{registered: map[string]string{}, trunks: map[string]string{"mail": "1", "odd": "2"},
nodes: map[string][]string{"mail": nil, "none": {"ace"}}, holder: "novox"},
keys: []delivered{{module: "mail"}, {module: "odd"}, {module: "none"}},
acts: []string{"catalogue", "take in mail at 1"},
want: map[string]want{"mail": {taken: true}, "odd": {}, "none": {}}},
{name: "a grant not written refuses the key, never swallowed",
mesh: aTakeIn{registered: map[string]string{}, trunks: map[string]string{"mail": "1"},
nodes: map[string][]string{"mail": {"ace"}}, holder: "novox", grantsErr: errors.New("the bucket mesh-controller_grants: not found")},
keys: []delivered{{module: "mail"}},
acts: []string{"catalogue", "take in mail at 1", "compose ace", "compose novox and grants"}, passes: 2,
want: map[string]want{"mail": {verdict: refused, err: true}}},
{name: "a node that fails refuses only its modules",
mesh: aTakeIn{registered: map[string]string{}, trunks: map[string]string{"mail": "1", "web": "2"},
nodes: map[string][]string{"mail": {"ace"}, "web": {"g14"}}, holder: "novox",
composeErr: map[string]error{"ace": errors.New("store refused")}},
keys: []delivered{{module: "mail"}, {module: "web"}},
acts: []string{"catalogue", "take in mail at 1", "take in web at 2", "compose ace,g14", "compose novox and grants"},
passes: 2, want: map[string]want{"mail": {verdict: refused, err: true}, "web": {taken: true}}},
{name: "waits and failures: a write's wait acknowledged, a read's refused, a failed key acknowledged",
mesh: aTakeIn{registered: map[string]string{}, trunks: map[string]string{"a": "1", "b": "2", "c": "3"},
nodes: map[string][]string{"a": {"ace"}, "b": {"ace"}, "c": {"ace"}}, holder: "novox",
takeInErr: map[string]error{"a": waitsForAWrite, "b": waitsOnARead, "c": keyFailed}},
keys: []delivered{{module: "a"}, {module: "b"}, {module: "c"}},
acts: []string{"catalogue", "take in a at 1", "take in b at 2", "take in c at 3"},
want: map[string]want{"a": {waits: true}, "b": {verdict: refused, waits: true}, "c": {err: true}}},
}
if !strings.Contains(said, "registered from the builder's done key") {
t.Errorf("said %q", said)
for _, c := range cases {
t.Run(c.name, func(t *testing.T) {
memo := c.memo
if memo == nil {
memo = map[string]string{}
}
settled, passes := c.mesh.steps().batch(t.Context(), c.keys, memo)
if !slices.Equal(c.mesh.acts, c.acts) {
t.Fatalf("the acts were\n %q\nwant\n %q", c.mesh.acts, c.acts)
}
if passes != c.passes {
t.Fatalf("%d composer pass(es), want %d", passes, c.passes)
}
if len(settled) != len(c.want) {
t.Fatalf("settled %d module(s), want %d: %+v", len(settled), len(c.want), settled)
}
for _, r := range settled {
w := c.want[r.module]
got := want{verdict: r.verdict, taken: r.taken, nothingNew: r.nothingNew, waits: r.waits != "", err: r.err != nil}
if got != w {
t.Errorf("%s: %+v, want %+v (%v)", r.module, got, w, r.err)
}
if r.taken && memo[r.module] == "" {
t.Errorf("%s was taken in and not remembered", r.module)
}
}
})
}
acts = nil
if _, err := stepsFor(&acts, "new", "new", fp, []string{"novox"}, "novox").take(context.Background(), "mesh-mail"); err != nil {
t.Fatal(err)
}
if got := strings.Join(acts, "; "); got != "compose novox; grants" {
t.Errorf("a module registered at its trunk already: %q", got)
}
// And each delivered key is acknowledged or refused by its module's verdict, the refused with the retry's delay.
t.Run("acknowledged and refused", func(t *testing.T) {
mesh := aTakeIn{registered: map[string]string{}, trunks: map[string]string{"a": "1", "b": "2"},
nodes: map[string][]string{"a": {"ace"}, "b": {"ace"}}, takeInErr: map[string]error{"b": waitsOnARead}}
var acked, naked []string
msg := func(module string) takeInMsg {
return takeInMsg{delivered: delivered{module: module}, ack: func() { acked = append(acked, module) },
nak: func(d time.Duration) {
if d != doneKeyRetry {
t.Errorf("refused with a delay of %s", d)
}
naked = append(naked, module)
}}
}
state := &takeInState{memo: map[string]string{}, raised: map[string]bool{}, saidWait: map[string]string{}}
said := state.take(t.Context(), mesh.steps(), []takeInMsg{msg("a"), msg("b"), msg("a")})
if !slices.Equal(acked, []string{"a", "a"}) || !slices.Equal(naked, []string{"b"}) {
t.Fatalf("acknowledged %v and refused %v", acked, naked)
}
if !strings.Contains(said, "1 taken in") || !strings.Contains(said, "1 waiting") {
t.Fatalf("said %q", said)
}
})
}
for name, s := range map[string]doneKeySteps{
"assigned nowhere": stepsFor(&acts, "old", "new", fp, nil, "novox"),
"no trunk value": stepsFor(&acts, "old", "new", "", []string{"ace"}, "novox"),
} {
acts = nil
if _, err := s.take(context.Background(), "mesh-mail"); err != nil {
t.Fatal(err)
}
if name == "assigned nowhere" && strings.Join(acts, "; ") != "take in mesh-mail at new" {
t.Errorf("%s: %v", name, acts)
}
if name == "no trunk value" && len(acts) != 0 {
t.Errorf("%s: %v", name, acts)
}
// A backlog that stands above zero with nothing delivered for five minutes is a stall; one that moves, or empties, is
// not (ADR 0544 §6).
func TestATakeInStallsWhenItsBacklogStandsStill(t *testing.T) {
at := time.Date(2026, 10, 25, 17, 36, 0, 0, time.UTC)
w := &stallWatch{}
read := func(pending uint64, delivered ...uint64) takeInBacklog {
return takeInBacklog{Pending: pending, delivered: delivered}
}
acts = nil
if _, err := stepsFor(&acts, "old", "new", fp, []string{"ace"}, "novox").take(context.Background(), "unknown"); err != nil || len(acts) != 0 {
t.Errorf("a module the catalogue does not hold: %v %v", acts, err)
steps := []struct {
after time.Duration
backlog takeInBacklog
stalled, changed bool
}{
{0, read(190, 10, 3), false, false},
{4 * time.Minute, read(190, 10, 3), false, false},
{5 * time.Minute, read(190, 10, 3), true, true},
{6 * time.Minute, read(190, 10, 3), true, false},
{7 * time.Minute, read(180, 11, 3), false, true},
{20 * time.Minute, read(0, 11, 3), false, false},
}
for i, s := range steps {
stalled, changed := w.look(s.backlog, at.Add(s.after))
if stalled != s.stalled || changed != s.changed {
t.Fatalf("reading %d: stalled %v changed %v, want %v %v", i, stalled, changed, s.stalled, s.changed)
}
}
}
@@ -115,15 +276,3 @@ func TestAChangedUserListIsANewRevisionOfTheBusModulesCompositionAlone(t *testin
}
}
}
// 1945 (the review's D1): a grant that is not written is the take-in's error, never swallowed: with no send left,
// nothing else would write it, and no want could name the fingerprint.
func TestAGrantNotWrittenIsTheTakeInsError(t *testing.T) {
fp := sdkdelivery.SourcePrefix2 + strings.Repeat("c", 64)
var acts []string
s := stepsFor(&acts, "old", "new", fp, []string{"ace"}, "novox")
s.grants = func(context.Context) error { return errors.New("the bucket mesh-controller_grants: not found") }
if _, err := s.take(context.Background(), "mesh-mail"); err == nil || !strings.Contains(err.Error(), "grants") {
t.Errorf("a grant not written: %v", err)
}
}
cmd/mesh-controller/composed_keys_waits_test.go
-184
View File
@@ -1,184 +0,0 @@
package main
import (
"context"
"fmt"
"strings"
"sync"
"testing"
"time"
)
// A take-in that waits is said once for each reason it waits for, and tried again every doneKeyRetry until it is taken
// in: never left for a key write that may not come (issue 1945, live on 2026-10-25: mesh-host's take-in at 26bd waited,
// said nothing, and nothing woke it until another write).
func TestAWaitingTakeInIsSaidOnceAndTriedAgain(t *testing.T) {
was, wasRetry, wasSay := doneKeyTakeOnce, doneKeyRetry, doneKeysSay
t.Cleanup(func() { doneKeyTakeOnce, doneKeyRetry, doneKeysSay = was, wasRetry, wasSay })
doneKeyRetry = 5 * time.Millisecond
var mu sync.Mutex
var said []string
doneKeysSay = func(s string) { mu.Lock(); said = append(said, s); mu.Unlock() }
calls := 0
doneKeyTakeOnce = func(context.Context, *stores, string) error {
calls++
switch {
case calls <= 2:
return fmt.Errorf("%w: the artifact store could not hold mesh-host's artifact", errKeyWaits)
case calls == 3:
return fmt.Errorf("%w: mesh-host's key cannot be read", errKeyWaits)
}
return nil
}
done := make(chan struct{})
go func() { takeDoneKey(context.Background(), nil, "mesh-host"); close(done) }()
select {
case <-done:
case <-time.After(5 * time.Second):
t.Fatal("a waiting take-in was not tried again until it was taken in")
}
if calls != 4 {
t.Fatalf("tried %d times, want 4", calls)
}
mu.Lock()
defer mu.Unlock()
waits := 0
for _, s := range said {
if strings.Contains(s, "mesh-host waits") {
waits++
}
}
if waits != 2 {
t.Fatalf("said %q: want each of the two reasons once", said)
}
}
// A wake while a take-in of the module already follows it adds no second one: the follower looks again itself.
func TestAWakeWhileATakeInFollowsAddsNone(t *testing.T) {
was, wasRetry, wasSay := doneKeyTakeOnce, doneKeyRetry, doneKeysSay
t.Cleanup(func() { doneKeyTakeOnce, doneKeyRetry, doneKeysSay = was, wasRetry, wasSay })
doneKeyRetry = 20 * time.Millisecond
doneKeysSay = func(string) {}
var mu sync.Mutex
calls := 0
release := make(chan struct{})
doneKeyTakeOnce = func(context.Context, *stores, string) error {
mu.Lock()
calls++
mu.Unlock()
select {
case <-release:
return nil
default:
return fmt.Errorf("%w: the key says building", errKeyWaits)
}
}
follower := make(chan struct{})
go func() { takeDoneKey(context.Background(), nil, "board"); close(follower) }()
time.Sleep(5 * time.Millisecond)
returned := make(chan struct{})
go func() { takeDoneKey(context.Background(), nil, "board"); close(returned) }()
select {
case <-returned:
case <-time.After(time.Second):
t.Fatal("a second wake waited for the follower")
}
close(release)
<-follower
}
// A wake while a take-in of the module follows it is not lost: the follower goes round once more after it succeeds, so
// a newer build's key written meanwhile is taken in (the review of issue 1945).
func TestAWakeDuringATakeInIsTakenAfterIt(t *testing.T) {
was, wasRetry, wasSay := doneKeyTakeOnce, doneKeyRetry, doneKeysSay
t.Cleanup(func() { doneKeyTakeOnce, doneKeyRetry, doneKeysSay = was, wasRetry, wasSay })
doneKeyRetry = 20 * time.Millisecond
doneKeysSay = func(string) {}
var mu sync.Mutex
calls := 0
inFirst := make(chan struct{})
release := make(chan struct{})
doneKeyTakeOnce = func(context.Context, *stores, string) error {
mu.Lock()
calls++
n := calls
mu.Unlock()
if n == 1 {
close(inFirst)
<-release
}
return nil
}
follower := make(chan struct{})
go func() { takeDoneKey(context.Background(), nil, "mesh-host"); close(follower) }()
<-inFirst
takeDoneKey(context.Background(), nil, "mesh-host") // the newer key's wake, while the first runs
close(release)
<-follower
mu.Lock()
defer mu.Unlock()
if calls != 2 {
t.Fatalf("taken %d times: the wake during the take-in was lost", calls)
}
}
// A wake that comes after the follower's last look but before it lets go of the module is not lost: the follower looks
// at the mark again once it has let go, and takes the module once more (the review of issue 1945).
func TestAWakeJustBeforeTheFollowerLetsGoIsTaken(t *testing.T) {
was, wasRetry, wasSay, wasHook := doneKeyTakeOnce, doneKeyRetry, doneKeysSay, doneKeysBeforeLetGo
t.Cleanup(func() {
doneKeyTakeOnce, doneKeyRetry, doneKeysSay, doneKeysBeforeLetGo = was, wasRetry, wasSay, wasHook
})
doneKeyRetry = 10 * time.Millisecond
doneKeysSay = func(string) {}
var mu sync.Mutex
calls := 0
doneKeyTakeOnce = func(context.Context, *stores, string) error {
mu.Lock()
calls++
mu.Unlock()
return nil
}
woke := false
doneKeysBeforeLetGo = func(module string) {
if !woke {
woke = true
takeDoneKey(context.Background(), nil, module) // fails its TryLock and marks the module
}
}
takeDoneKey(context.Background(), nil, "mesh-host")
mu.Lock()
defer mu.Unlock()
if calls != 2 {
t.Fatalf("taken %d times: the wake just before the follower let go was lost", calls)
}
}
// A wake during the start's take-in of a module is not lost: the start pass reads the mark once it lets go of the module,
// and hands the module to takeDoneKey, which takes it once more (the review of issue 1945: the third dropped wake of its
// kind).
func TestAWakeDuringTheStartsTakeInIsTaken(t *testing.T) {
was, wasRetry, wasSay := doneKeyTakeOnce, doneKeyRetry, doneKeysSay
t.Cleanup(func() { doneKeyTakeOnce, doneKeyRetry, doneKeysSay = was, wasRetry, wasSay })
doneKeyRetry = 10 * time.Millisecond
doneKeysSay = func(string) {}
var mu sync.Mutex
calls := 0
doneKeyTakeOnce = func(context.Context, *stores, string) error {
mu.Lock()
calls++
n := calls
mu.Unlock()
if n == 1 {
takeDoneKey(context.Background(), nil, "mesh-host") // the newer key's wake, during the start's take-in
}
return nil
}
takeDoneKeyAtStart(context.Background(), nil, "mesh-host")
background.Wait()
mu.Lock()
defer mu.Unlock()
if calls != 2 {
t.Fatalf("taken %d times: the wake during the start's take-in was lost", calls)
}
}
cmd/mesh-controller/composer.go
+32 -10
View File
@@ -29,7 +29,13 @@ type composer struct {
type waiter struct {
nodes []string
grants bool
done chan error
done chan passResult
}
// passResult is a waiter's answer: the error of each of its nodes that failed, and of the grants when it asked for them.
type passResult struct {
failed map[string]error
grants error
}
func newComposer(compose func(context.Context, []string) map[string]error, grants func(context.Context) error) *composer {
@@ -39,7 +45,23 @@ func newComposer(compose func(context.Context, []string) map[string]error, grant
// compose asks for the nodes to be composed, and the grants written after when grants is true, and waits for the pass
// that does it, with the errors of its own nodes alone.
func (c *composer) compose(ctx context.Context, nodes []string, grants bool) error {
done := make(chan error, 1)
failed, grantsErr, err := c.composeEach(ctx, nodes, grants)
if err != nil {
return err
}
var errs []error
for _, n := range nodes {
if e := failed[n]; e != nil {
errs = append(errs, e)
}
}
return errors.Join(append(errs, grantsErr)...)
}
// composeEach is compose answering each node's error apart, and the grants' (ADR 0544 §5: a batch of take-ins asks one
// pass and settles each module by its own nodes); err is ctx's when it ends first.
func (c *composer) composeEach(ctx context.Context, nodes []string, grants bool) (map[string]error, error, error) {
done := make(chan passResult, 1)
c.mu.Lock()
for _, n := range nodes {
c.pending[n] = true
@@ -52,10 +74,10 @@ func (c *composer) compose(ctx context.Context, nodes []string, grants bool) err
default:
}
select {
case err := <-done:
return err
case r := <-done:
return r.failed, r.grants, nil
case <-ctx.Done():
return ctx.Err()
return nil, nil, ctx.Err()
}
}
@@ -90,16 +112,16 @@ func (c *composer) run(ctx context.Context) {
grantsErr = c.writeGrants(ctx)
}
for _, w := range waiters {
var errs []error
r := passResult{failed: map[string]error{}}
for _, n := range w.nodes {
if err := failed[n]; err != nil {
errs = append(errs, err)
r.failed[n] = err
}
}
if w.grants && grantsErr != nil {
errs = append(errs, grantsErr)
if w.grants {
r.grants = grantsErr
}
w.done <- errors.Join(errs...)
w.done <- r
}
}
}
cmd/mesh-controller/key_takein.go
+11 -3
View File
@@ -39,6 +39,11 @@ import (
// errKeyWaits says a take-in from a key waits: for the builder, a reference, the store or the forge. Never a failure.
var errKeyWaits = errors.New("waits")
// errWaitsForAWrite is a take-in that waits for the builder or the indexer to write again (the key not done, its
// artifacts not referenced, no fingerprint of the new form): that write wakes it, so nothing looks again before it
// (ADR 0544 §4).
var errWaitsForAWrite = fmt.Errorf("%w for the builder's or the indexer's next write", errKeyWaits)
// errKeyFailed says the module's key failed: the builder's own words follow.
var errKeyFailed = errors.New("the builder's key failed")
@@ -298,12 +303,15 @@ func buildOfKey(ctx context.Context, keys keyReads, t keyTake, kc keyContext) (
waits := func(format string, a ...any) (inventory.Build, catalogue.Manifest, error) {
return inventory.Build{}, catalogue.Manifest{}, fmt.Errorf("%w: "+format, append([]any{errKeyWaits}, a...)...)
}
waitsForAWrite := func(format string, a ...any) (inventory.Build, catalogue.Manifest, error) {
return inventory.Build{}, catalogue.Manifest{}, fmt.Errorf("%w: "+format, append([]any{errWaitsForAWrite}, a...)...)
}
if t.Trunk.FP2 == "" && t.Hand == nil {
why := t.Trunk.None2
if why == "" {
why = "the indexer wrote no fingerprint of the new form"
}
return waits("%s at %.8s has no fingerprint to take a build by: %s", t.Module, t.Trunk.Commit, why)
return waitsForAWrite("%s at %.8s has no fingerprint to take a build by: %s", t.Module, t.Trunk.Commit, why)
}
k, err := keys.Build(ctx, t.Module, t.Trunk.FP2)
if t.Hand != nil {
@@ -319,9 +327,9 @@ func buildOfKey(ctx context.Context, keys keyReads, t keyTake, kc keyContext) (
case k.Failed():
return inventory.Build{}, catalogue.Manifest{}, fmt.Errorf("%w: %s", errKeyFailed, k.Says())
case !k.Done():
return waits("%s", k.Says())
return waitsForAWrite("%s", k.Says())
case len(k.Unreferenced()) > 0:
return waits("%s", k.Says())
return waitsForAWrite("%s", k.Says())
}
src, err := keys.Manifest(ctx, k.Build.Manifest)
if err != nil {
cmd/mesh-controller/plain_words.go
+7
View File
@@ -70,6 +70,13 @@ var plainWordings = map[string]func(conditions.Observation) words{
"none can be given it yet. It is tried again every half minute; the details say what fails.", m),
Resolved: "the new build of " + m + " can be delivered"}
}),
// The take-in of done keys whose backlog stood still (take_in_consumer.go, novox/hq ADR 0544 §6).
kindTakeInStalled: worded(func(o conditions.Observation) words {
return words{Headline: "New builds are not being prepared for the machines",
Explanation: "The mesh prepares each new build for the machines as it is made, and has prepared none for " +
"five minutes while builds wait. Until it moves, no new build reaches a machine; the details say how many wait.",
Resolved: "new builds are being prepared again"}
}),
// Machines and their node-engine.
"silent": worded(func(o conditions.Observation) words {
m := machineOr(o, "a machine")
cmd/mesh-controller/readable.go
+4
View File
@@ -111,6 +111,9 @@ type meshStatus struct {
// why they could not be read.
Pending []pendingStatus `json:"pending,omitempty"`
PendingUnread string `json:"pendingUnread,omitempty"`
// TakeIn is the take-in's backlog of done keys: keys not delivered yet and keys being taken in (novox/hq ADR 0544
// §6). Absent where it cannot be read.
TakeIn *takeInBacklog `json:"take-in,omitempty"`
}
// pendingStatus is one pending assignment as status says it.
@@ -266,6 +269,7 @@ func statusAsJSON(asked answers) ([]byte, error) {
out.Failing = providerStandings(asked.conditions)
out.Overflowing = asked.overflowing
out.PendingUnread = asked.pendingUnread
out.TakeIn = asked.takeIn
for _, p := range asked.pending {
out.Pending = append(out.Pending, pendingStatus{Node: p.Node, Module: p.Module, State: p.State,
Build: p.Build, Repository: p.Repository, Path: p.Path, Since: p.Since, Settled: p.Settled, Note: p.Note})
cmd/mesh-controller/status.go
+11
View File
@@ -151,6 +151,10 @@ func printStatus(asked answers) error {
}
printPending(asked.pending, asked.pendingUnread)
if b := asked.takeIn; b != nil && b.Total() > 0 {
fmt.Printf("the take-in of done keys has %d key(s) not taken in yet: %d waiting to be delivered, %d being taken in\n\n",
b.Total(), b.Pending, b.Taking)
}
if len(behind) > 0 {
var names []string
@@ -471,6 +475,13 @@ func theThreeQuestions(ctx context.Context, open *stores) (answers, error) {
out.pendingUnread = err.Error()
err = nil
}
// And the take-in's backlog of done keys, read from its consumers on the bus (novox/hq ADR 0544 §6): where there is
// a bus; a stall is a condition of its own, so this is said and judges nothing.
if _, onBus := broker.BusAddress(); onBus == nil {
if b, berr := readTakeInBacklog(ctx); berr == nil {
out.takeIn = &b
}
}
// And how many repairs were done by hand this week (novox/hq to-be 45 §7) — where there is a bus
// to read the log from; a process with none has no log to count.
if _, onBus := broker.BusAddress(); onBus == nil {
cmd/mesh-controller/take_in_consumer.go
+365
View File
@@ -0,0 +1,365 @@
package main
import (
"context"
"errors"
"fmt"
"strings"
"sync"
"time"
sdkdelivery "git.novox.be/novox/mesh-sdk/go/delivery"
"github.com/nats-io/nats.go/jetstream"
"github.com/novox/mesh-controller/internal/broker"
"github.com/novox/mesh-controller/internal/conditions"
)
// The take-in's two durable pull consumers (novox/hq ADR 0544; research 072, milestones 2 and 4). On the indexer's
// modules bucket the consumer delivers each module's last trunk value when it is made, then each write: the start's look
// of a controller whose consumer is new, and nothing at a restart but what the controller before it left
// unacknowledged. On the builder's builds bucket it delivers each write from when it is made. A key is acknowledged once
// its module is taken in, has nothing new, or waits for a write that will be delivered; any other wait or failure is
// refused with a delay, and the bus delivers it again. At most startTakeIns keys of each are unacknowledged: the bound on
// the store (issues 1945 and 1277: a burst of take-ins once filled it until it refused every connection).
// startTakeIns is how many keys each consumer leaves unacknowledged at most: the take-ins that may wait on the store.
const startTakeIns = 16
// takeInAckWait is how long a delivered key may stay unacknowledged before the bus delivers it again: longer than a batch
// takes, its two composer passes among it.
const takeInAckWait = 10 * time.Minute
// takeInGather is how long a batch waits for more keys after its first, so a burst is one batch.
var takeInGather = 200 * time.Millisecond
// takeInStall is how long a backlog may stand without a key delivered before it is a condition (ADR 0544 §6).
var takeInStall = 5 * time.Minute
// takeInMsg is one delivered key, as the take-in uses it.
type takeInMsg struct {
delivered
ack func()
nak func(delay time.Duration)
}
// takeInKeyOf is the module a delivered key names, false for a delete or a purge, which takes nothing in.
func takeInKeyOf(bucket string, m jetstream.Msg) (delivered, bool) {
if op := m.Headers().Get("KV-Operation"); op == "DEL" || op == "PURGE" {
return delivered{}, false
}
key := strings.TrimPrefix(m.Subject(), "$KV."+bucket+".")
module, _, _ := strings.Cut(key, ".")
if module == "" {
return delivered{}, false
}
again := false
if meta, err := m.Metadata(); err == nil {
again = meta.NumDelivered > 1
}
return delivered{module: module, again: again}, true
}
// takeInConsumer is one bucket's consumer of the take-in.
type takeInConsumer struct {
bucket string
consumer jetstream.Consumer
}
// takeInConsumers makes or finds the take-in's consumer on each bucket's stream; one that cannot be made is said, and
// the other is still read.
func takeInConsumers(ctx context.Context, js jetstream.JetStream) []takeInConsumer {
var out []takeInConsumer
for _, bucket := range broker.TakeInBuckets {
deliver := jetstream.DeliverNewPolicy
if bucket == sdkdelivery.ModulesBucket {
deliver = jetstream.DeliverLastPerSubjectPolicy // the start's look: every module's trunk value once
}
c, err := js.CreateOrUpdateConsumer(ctx, "KV_"+bucket, jetstream.ConsumerConfig{
Durable: broker.TakeInConsumer, FilterSubject: "$KV." + bucket + ".>", DeliverPolicy: deliver,
AckPolicy: jetstream.AckExplicitPolicy, AckWait: takeInAckWait, MaxAckPending: startTakeIns,
Description: "the controller's take-in of done keys (novox/hq ADR 0544)",
})
if err != nil {
fmt.Printf("done keys: the take-in's consumer on %s is not made, so its keys are not taken in until the "+
"next start: %v\n", bucket, err)
continue
}
out = append(out, takeInConsumer{bucket: bucket, consumer: c})
}
return out
}
// watchDoneKeys takes in the done keys the two consumers deliver, batch after batch, until ctx ends, and watches their
// backlog. A consumer that cannot be made is said: its bucket is then not taken in until the next start.
func watchDoneKeys(ctx context.Context, open *stores) {
keys, closeKeys, err := openBuilderKeys()
if err != nil {
fmt.Printf("done keys: the builder's keys are not taken in: %v\n", err)
return
}
consumers := takeInConsumers(ctx, keys.JS)
if len(consumers) == 0 {
closeKeys()
return
}
in := make(chan takeInMsg)
read := make([]jetstream.Consumer, 0, len(consumers))
for _, c := range consumers {
bucket := c.bucket
read = append(read, c.consumer)
it, err := c.consumer.Messages(jetstream.PullMaxMessages(startTakeIns))
if err != nil {
fmt.Printf("done keys: the take-in's consumer on %s is not read: %v\n", bucket, err)
continue
}
goServing(ctx, func() {
<-ctx.Done()
it.Stop()
})
goServing(ctx, func() {
for {
m, err := it.Next()
if err != nil {
if ctx.Err() == nil && !errors.Is(err, jetstream.ErrMsgIteratorClosed) {
fmt.Printf("done keys: the take-in's consumer on %s stopped: %v\n", bucket, err)
}
return
}
d, ok := takeInKeyOf(bucket, m)
if !ok {
_ = m.Ack()
continue
}
msg := takeInMsg{delivered: d, ack: func() { _ = m.Ack() },
nak: func(delay time.Duration) { _ = m.NakWithDelay(delay) }}
select {
case in <- msg:
case <-ctx.Done():
return
}
}
})
}
backlog := func(ctx context.Context) (takeInBacklog, error) { return backlogOf(ctx, read) }
steps := liveTakeInSteps(open, keys, keys.JS)
goServing(ctx, func() { takeInLoop(ctx, in, steps, backlog) })
goServing(ctx, func() { watchTakeInStall(ctx, backlog) })
goServing(ctx, func() {
<-ctx.Done()
closeKeys()
})
}
// takeInLoop gathers delivered keys into batches and takes each batch in, until ctx ends.
func takeInLoop(ctx context.Context, in <-chan takeInMsg, steps takeInSteps,
backlog func(context.Context) (takeInBacklog, error)) {
t := &takeInState{memo: map[string]string{}, raised: map[string]bool{}, saidWait: map[string]string{}}
for {
var batch []takeInMsg
select {
case <-ctx.Done():
return
case m := <-in:
batch = append(batch, m)
}
gather := time.NewTimer(takeInGather)
more:
for len(batch) < 2*startTakeIns {
select {
case m := <-in:
batch = append(batch, m)
case <-gather.C:
break more
case <-ctx.Done():
gather.Stop()
return
}
}
gather.Stop()
said := t.take(ctx, steps, batch)
if b, err := backlog(ctx); err == nil {
said += fmt.Sprintf("; backlog %d", b.Total())
}
doneKeysSay("done keys: " + said)
}
}
// takeInState is what the take-in keeps between batches in this process: the trunk each module was last taken in at,
// the modules whose condition it raised, and the wait last said of each.
type takeInState struct {
memo map[string]string
raised map[string]bool
saidWait map[string]string
}
// take takes one batch in and acknowledges or refuses each key, raising and clearing the modules' conditions. It answers
// the batch in a line.
func (t *takeInState) take(ctx context.Context, steps takeInSteps, batch []takeInMsg) string {
keys := make([]delivered, len(batch))
for i, m := range batch {
keys[i] = m.delivered
}
settled, passes := steps.batch(ctx, keys, t.memo)
verdicts := make(map[string]verdict, len(settled))
var taken, nothing, waiting, failed int
for _, r := range settled {
verdicts[r.module] = r.verdict
o := doneKeyObservation(r.module)
switch {
case r.err != nil:
failed++
doneKeysSay(fmt.Sprintf("done keys: %s: %v", r.module, r.err))
if keeper := conditionsFrom; keeper != nil && !t.raised[r.module] && ctx.Err() == nil &&
!errors.Is(r.err, context.Canceled) {
o.Summary += r.err.Error()
if _, err := keeper.Observe(ctx, o); err == nil {
t.raised[r.module] = true
}
}
case r.waits != "":
waiting++
if t.saidWait[r.module] != r.waits {
doneKeysSay(fmt.Sprintf("done keys: %s %s", r.module, r.waits))
t.saidWait[r.module] = r.waits
}
case r.taken:
taken++
delete(t.saidWait, r.module)
if r.said != "" {
doneKeysSay("done keys: " + r.said)
}
// Cleared on every take-in, whoever raised it: a controller before a restart, whose key was delivered again.
if keeper := conditionsFrom; keeper != nil {
if _, err := keeper.Clear(ctx, o.Key(), "composed and granted"); err == nil {
delete(t.raised, r.module)
}
}
case r.nothingNew:
nothing++
}
}
for _, m := range batch {
if verdicts[m.module] == refused {
m.nak(doneKeyRetry)
} else {
m.ack()
}
}
return fmt.Sprintf("%d key(s) of %d module(s): %d taken in, %d with nothing new, %d waiting, %d failed; %d composer pass(es)",
len(batch), len(settled), taken, nothing, waiting, failed, passes)
}
// takeInBacklog is the take-in's backlog on the bus: keys not delivered yet, keys delivered and not acknowledged, and
// the last delivered sequence of each consumer, which moves when anything is delivered.
type takeInBacklog struct {
Pending uint64 `json:"pending"`
Taking uint64 `json:"taking"`
delivered []uint64
}
// Total is every key the take-in has not acknowledged.
func (b takeInBacklog) Total() uint64 { return b.Pending + b.Taking }
// backlogOf reads the consumers' own counts from the bus.
func backlogOf(ctx context.Context, consumers []jetstream.Consumer) (takeInBacklog, error) {
var b takeInBacklog
for _, c := range consumers {
info, err := c.Info(ctx)
if err != nil {
return takeInBacklog{}, err
}
b.Pending += info.NumPending
b.Taking += uint64(info.NumAckPending)
b.delivered = append(b.delivered, info.Delivered.Consumer)
}
return b, nil
}
// readTakeInBacklog is the take-in's backlog for a reader in any process (`status`): the consumers asked by name.
func readTakeInBacklog(ctx context.Context) (takeInBacklog, error) {
keys, closeKeys, err := openBuilderKeys()
if err != nil {
return takeInBacklog{}, err
}
defer closeKeys()
var consumers []jetstream.Consumer
for _, bucket := range broker.TakeInBuckets {
c, err := keys.JS.Consumer(ctx, "KV_"+bucket, broker.TakeInConsumer)
if err != nil {
return takeInBacklog{}, fmt.Errorf("the take-in's consumer on %s: %w", bucket, err)
}
consumers = append(consumers, c)
}
return backlogOf(ctx, consumers)
}
// kindTakeInStalled is the condition of a take-in whose backlog stood still (ADR 0544 §6).
const kindTakeInStalled = "take-in-stalled"
func takeInStalledObservation(b takeInBacklog, since time.Time) conditions.Observation {
return conditions.Observation{Scope: conditions.ScopeCore, ID: "take-in", Token: kindTakeInStalled,
Kind: kindTakeInStalled, Severity: conditions.Urgent, Resolver: conditions.ResolverOperator, Source: "done keys",
Summary: fmt.Sprintf("the take-in of done keys has %d key(s) waiting and has taken none in since %s",
b.Total(), since.UTC().Format(time.RFC3339))}
}
// stallWatch judges the backlog read at each look: stalled once it stood above zero, with nothing delivered, for
// takeInStall.
type stallWatch struct {
mu sync.Mutex
last []uint64
movedAt time.Time
stalled bool
lastTotal uint64
}
// look takes a reading at now and answers whether the take-in stalls and whether that changed.
func (w *stallWatch) look(b takeInBacklog, now time.Time) (stalled, changed bool) {
w.mu.Lock()
defer w.mu.Unlock()
moved := b.Total() == 0 || len(w.last) != len(b.delivered)
for i := range b.delivered {
if !moved && b.delivered[i] != w.last[i] {
moved = true
}
}
if moved || w.movedAt.IsZero() {
w.movedAt = now
}
w.last, w.lastTotal = b.delivered, b.Total()
nowStalled := b.Total() > 0 && now.Sub(w.movedAt) >= takeInStall
changed = nowStalled != w.stalled
w.stalled = nowStalled
return nowStalled, changed
}
// watchTakeInStall reads the backlog every minute and raises take-in-stalled while it stands still, clearing it when it
// moves: a watchdog's timer, as research 072 §2 keeps them until the health part leaves the controller.
func watchTakeInStall(ctx context.Context, backlog func(context.Context) (takeInBacklog, error)) {
w := &stallWatch{}
tick := time.NewTicker(time.Minute)
defer tick.Stop()
for {
select {
case <-ctx.Done():
return
case <-tick.C:
}
b, err := backlog(ctx)
if err != nil {
continue
}
stalled, changed := w.look(b, time.Now())
keeper := conditionsFrom
if !changed || keeper == nil {
continue
}
o := takeInStalledObservation(b, w.movedAt)
if stalled {
doneKeysSay("done keys: " + o.Summary)
_, _ = keeper.Observe(ctx, o)
} else {
_, _ = keeper.Clear(ctx, o.Key(), "the take-in moves again")
}
}
}
internal/broker/composed.go
+8
View File
@@ -182,3 +182,11 @@ func RaiseAfterDeclarations(r Raiser) error {
}
return AssertMeshConsumers(r)
}
// TakeInConsumer is the durable consumer the controller takes done keys in from, on the stream of each of TakeInBuckets
// (novox/hq ADR 0544 §1).
const TakeInConsumer = "mesh-controller-take-in"
// TakeInBuckets are the buckets the take-in consumes: the indexer's modules (each module's trunk value) and the builder's
// builds (each build's key).
var TakeInBuckets = []string{delivery.ModulesBucket, delivery.BuilderBuildsBucket}
internal/broker/nats.go
+15
View File
@@ -432,6 +432,16 @@ var DegradableGrants = []DegradableGrant{
Reason: "carried only at genesis: the controller composes its own permissions into the bus module's composition, " +
"and writes a composition's digest only after the bus applied it (novox/hq ADR 0539 clauses 2 and 7)",
},
{
Subject: "$JS.ACK.KV_" + delivery.ModulesBucket + "." + TakeInConsumer + ".>",
Reason: "refused: an acknowledgement refused leaves the key delivered again after its wait, and the take-in's first " +
"batch with something new composes the bus's node, which grants it (novox/hq ADR 0544 §1)",
},
{
Subject: "$JS.ACK.KV_" + delivery.BuilderBuildsBucket + "." + TakeInConsumer + ".>",
Reason: "refused: an acknowledgement refused leaves the key delivered again after its wait, and the take-in's first " +
"batch with something new composes the bus's node, which grants it (novox/hq ADR 0544 §1)",
},
{
Subject: "$KV." + GrantsBucket + ".>",
Reason: "carried only at genesis: the controller composes its own permissions and pushes them to the bus at its " +
@@ -707,6 +717,11 @@ func PermissionsFor(p Principal) (Permissions, error) {
// ever, refused by the list it already has.
sub = append(sub, ControllerFollows...)
pub = append(pub, "$JS.ACK.EVENTS."+ControllerName+".>", "$JS.ACK.EVENTS."+MergesConsumerName+".>")
// And the take-in's two durable consumers on the indexer's and the builder's buckets (novox/hq ADR 0544 §1):
// acknowledged after each key is taken in, and no other consumer of those streams.
for _, bucket := range TakeInBuckets {
pub = append(pub, "$JS.ACK.KV_"+bucket+"."+TakeInConsumer+".>")
}
// **Where an enrolment's answer goes**, and `allow_responses` does not cover it. That
// permits one reply to the reply subject of a message the user received — and a message a
internal/broker/testdata/composed.conf
Vendored
+1 -1
View File
@@ -29,7 +29,7 @@ accounts {
]
users = [
{ user: "controller", password: "$2a$11$cccccccccccccccccccccc", permissions: {
publish: { allow: ["$JS.ACK.CONTROL.controller.>", "$JS.ACK.DEAD_LETTER_NOTICES.controller.>", "$JS.ACK.EVENTS.controller-merges.>", "$JS.ACK.EVENTS.controller.>", "$JS.API.>", "$KV.mesh-controller_asked.>", "$KV.mesh-controller_assignments.>", "$KV.mesh-controller_builds.>", "$KV.mesh-controller_calls.>", "$KV.mesh-controller_composed.>", "$KV.mesh-controller_condition-history.>", "$KV.mesh-controller_conditions.>", "$KV.mesh-controller_declared-config.>", "$KV.mesh-controller_declared-packages.>", "$KV.mesh-controller_delivery-merges.>", "$KV.mesh-controller_grants.>", "$KV.mesh-controller_hand-acts.>", "$KV.mesh-controller_lease.>", "$KV.mesh-controller_predefined.>", "$KV.mesh-controller_rollouts.>", "$KV.mesh-controller_states-announced.>", "$SRV.INFO", "_INBOX.enrol.>", "mesh.again.>", "mesh.assignment.>", "mesh.events.dead.>", "mesh.mod.*.tool.>", "mesh.mod.mesh-controller.event.states", "mesh.node.>", "mesh.seat.mesh-controller.event.applied", "mesh.seat.mesh-controller.event.artifacts-gone", "mesh.seat.mesh-controller.event.assignment-changed", "mesh.seat.mesh-controller.event.bucket-deleted", "mesh.seat.mesh-controller.event.built-before", "mesh.seat.mesh-controller.event.checked", "mesh.seat.mesh-controller.event.condition-changed", "mesh.seat.mesh-controller.event.condition-cleared", "mesh.seat.mesh-controller.event.condition-raised", "mesh.seat.mesh-controller.event.declared-config-changed", "mesh.seat.mesh-controller.event.doctor-heartbeat", "mesh.seat.mesh-controller.event.healer-acted", "mesh.seat.mesh-controller.event.predefined-changed", "mesh.seat.mesh-controller.event.refused", "mesh.seat.mesh-controller.event.rolled-back", "mesh.seat.mesh-controller.event.secret-replaced", "mesh.seat.mesh-controller.tool.plans", "mesh.seat.mesh-controller.tool.push", "mesh.seat.mesh-controller.tool.upgrade", "mesh.seat.mesh-delivery.tool.close", "mesh.seat.mesh-delivery.tool.groups", "mesh.seat.mesh-delivery.tool.release", "mesh.seat.mesh-delivery.tool.stalled", "mesh.seat.mesh-delivery.tool.stop", "mesh.seat.mesh-delivery.tool.times", "mesh.seat.node-backup.tool.backed-up.*", "mesh.seat.node-backup.tool.now.*", "mesh.seat.node-intrusion-prevention.tool.banned.*", "mesh.seat.node-launcher.tool.secret.*", "mesh.seat.node-service-manager.tool.restart.*", "mesh.seat.operator-channel.tool.tell"] }
publish: { allow: ["$JS.ACK.CONTROL.controller.>", "$JS.ACK.DEAD_LETTER_NOTICES.controller.>", "$JS.ACK.EVENTS.controller-merges.>", "$JS.ACK.EVENTS.controller.>", "$JS.ACK.KV_mesh-delivery-builder_builds.mesh-controller-take-in.>", "$JS.ACK.KV_mesh-delivery-indexer_modules.mesh-controller-take-in.>", "$JS.API.>", "$KV.mesh-controller_asked.>", "$KV.mesh-controller_assignments.>", "$KV.mesh-controller_builds.>", "$KV.mesh-controller_calls.>", "$KV.mesh-controller_composed.>", "$KV.mesh-controller_condition-history.>", "$KV.mesh-controller_conditions.>", "$KV.mesh-controller_declared-config.>", "$KV.mesh-controller_declared-packages.>", "$KV.mesh-controller_delivery-merges.>", "$KV.mesh-controller_grants.>", "$KV.mesh-controller_hand-acts.>", "$KV.mesh-controller_lease.>", "$KV.mesh-controller_predefined.>", "$KV.mesh-controller_rollouts.>", "$KV.mesh-controller_states-announced.>", "$SRV.INFO", "_INBOX.enrol.>", "mesh.again.>", "mesh.assignment.>", "mesh.events.dead.>", "mesh.mod.*.tool.>", "mesh.mod.mesh-controller.event.states", "mesh.node.>", "mesh.seat.mesh-controller.event.applied", "mesh.seat.mesh-controller.event.artifacts-gone", "mesh.seat.mesh-controller.event.assignment-changed", "mesh.seat.mesh-controller.event.bucket-deleted", "mesh.seat.mesh-controller.event.built-before", "mesh.seat.mesh-controller.event.checked", "mesh.seat.mesh-controller.event.condition-changed", "mesh.seat.mesh-controller.event.condition-cleared", "mesh.seat.mesh-controller.event.condition-raised", "mesh.seat.mesh-controller.event.declared-config-changed", "mesh.seat.mesh-controller.event.doctor-heartbeat", "mesh.seat.mesh-controller.event.healer-acted", "mesh.seat.mesh-controller.event.predefined-changed", "mesh.seat.mesh-controller.event.refused", "mesh.seat.mesh-controller.event.rolled-back", "mesh.seat.mesh-controller.event.secret-replaced", "mesh.seat.mesh-controller.tool.plans", "mesh.seat.mesh-controller.tool.push", "mesh.seat.mesh-controller.tool.upgrade", "mesh.seat.mesh-delivery.tool.close", "mesh.seat.mesh-delivery.tool.groups", "mesh.seat.mesh-delivery.tool.release", "mesh.seat.mesh-delivery.tool.stalled", "mesh.seat.mesh-delivery.tool.stop", "mesh.seat.mesh-delivery.tool.times", "mesh.seat.node-backup.tool.backed-up.*", "mesh.seat.node-backup.tool.now.*", "mesh.seat.node-intrusion-prevention.tool.banned.*", "mesh.seat.node-launcher.tool.secret.*", "mesh.seat.node-service-manager.tool.restart.*", "mesh.seat.operator-channel.tool.tell"] }
subscribe: { allow: ["$JS.API.>", "$JS.EVENT.ADVISORY.CONSUMER.DELETED.>", "$JS.EVENT.ADVISORY.CONSUMER.MAX_DELIVERIES.>", "$SRV.INFO", "$SRV.INFO.mesh-controller", "$SRV.INFO.mesh-controller.>", "$SRV.PING", "$SRV.PING.mesh-controller", "$SRV.PING.mesh-controller.>", "$SRV.STATS", "$SRV.STATS.mesh-controller", "$SRV.STATS.mesh-controller.>", "_DELIVER.controller", "_DELIVER.controller.>", "_INBOX.controller.>", "mesh.control.>", "mesh.mod.*.event.provisioner.failing", "mesh.mod.*.event.provisioner.recovered", "mesh.mod.*.event.provisioner.retirement", "mesh.mod.gitea.event.pull.merged", "mesh.mod.gitea.event.pull.updated", "mesh.mod.mesh-catalog.event.catching-up", "mesh.mod.mesh-catalog.event.upgraded", "mesh.seat.mesh-build-machine.event.built", "mesh.seat.mesh-controller.tool.>", "mesh.seat.node-build-agent.event.built", "mesh.seat.operator-channel.event.decided.mesh-controller"] }
allow_responses: { max: 1, ttl: "1m" }
} }