Compare commits

...
Author SHA1 Message Date
jschoubben b980ba3631 Raise no condition for a take-in, a layout or a put cut by a cancellation (issue 1945)
Thirty urgent fingerprint-not-composed conditions were live on 2026-10-25, each a
done key's take-in a stop or a restart cut with "context canceled". A
cancellation is no fault of the build, the chain or the bucket: a take-in or a
layout whose controller stops returns and the next one does it; one whose own
step was cancelled is tried again without a condition; a send cut by a
cancellation counts as no failure to put its files.

The put test now starts and leaves the node's failure count at nothing, so a
second run of it (-count=2) finds no count left by the first.
2026-10-25 16:29:56 +01:00
jschoubben e092f01e54 End every goroutine serve starts before serve returns, so no loop outlives its controller (issue 1945)
The try of the on-change mirror failed with every test passed: goroutines a
served test started outlived its serve and met the next test's globals and its
closed stores. Under -race the served tests show it as a race on conditionsFrom
(watchTheMesh writing it while settlingPending, already running, read it), and
a take-in cut by the stop raised fingerprint-not-composed, a kind with no plain
words, which fails the package after its PASS line.

- serve counts every goroutine it and its parts start (serving_group.go) and
  ends and waits for them before anything else is undone: the lease's watch,
  the pending settler, the status summary, the facts, the mirrors and the new
  declared-packages keeper with its two bucket watches, the declaration sweep,
  the read-call counter, the watches' closers, the have watches, the vault move
  on a report, and the watchdogs, doctor, healers and asker under their own
  group.
- The condition store opens before anything that reads conditionsFrom starts;
  askerFrom is an atomic pointer, read from the keeper's goroutines.
- A done key's take-in cut by the stop raises nothing, and fingerprint-not-composed
  has plain words.
2026-10-25 16:03:46 +01:00
jschoubben b6379b9e86 Compose the declared-packages mirror only when what it composes from changed (issue 1945)
The mirror composed every node in full on the record mirrors' 10 s tick; a
pass takes about 45 s on the control node, so the controller never rested
(130-175 % CPU, issue 1945 note 22). It now passes on a watch of the
compositions' digests and of the assignments, one pass at a time with the
wakes of a pass coalesced into one after it (at least 30 s apart), skips the
composition when the stores' Input digest is unchanged, and keeps a safety
pass every 10 minutes that composes nothing when nothing changed.
2026-10-25 15:07:32 +01:00
mesh-tools 2bcd85d2df Merge pull request 'Never put a module whose want names its own composition into the node's own, and read a re-registered module's fingerprint from the indexer (issue 1945)' (#461) from fix/r6-own-composition-dupes into main 2026-10-25 13:37:34 +00:00
jschoubben ac34bce49f Take a module's fingerprint from the indexer when its commit has no build, and never compose a module its want names into the node's own (issue 1945)
mesh/check pass ac34bce49f58 on 2bcd85d2df4c | unit 966 run, 171 kept, 2 skipped | integration 1625 run, 232 kept, 7 skipped | over bound: integration…
mesh/delivery merged: source.main moved to novox/mesh-controller@2bcd85d2df4c
A catalogue merge that leaves a module's sources alone registers it at the new
commit from the same done key, whose build record stands at its first commit,
so the commit had no build and the module no fingerprint of the new form: it
went into the node's own composition while its want named its own, and every
node declared its files twice. currentFingerprints now reads the indexer's
trunk value at the commit when no build points at it (1945/AC15), and a module
a want or have names by its own composition is left out of the node's own
(1945/AC16).
2026-10-25 14:10:51 +01:00
mesh-tools af2b0844a1 Merge pull request 'Compose each machine once per pass, so take-ins of done builds stop colliding and every new build gets its composition and credential (issue 1945)' (#460) from fix/r6-compose-once into main 2026-10-25 12:30:32 +00:00
27 changed files with 1067 additions and 97 deletions
+4 -3
View File
@@ -12,6 +12,7 @@ import (
"sort"
"strings"
"sync"
"sync/atomic"
"time"
"github.com/nats-io/nats.go"
@@ -27,7 +28,7 @@ import (
)
// askerFrom is the serving controller's asker; nil in any other process.
var askerFrom *asker
var askerFrom atomic.Pointer[asker]
// askWithin is how long a verb a warrant chose is given to answer.
const askWithin = time.Minute
@@ -573,8 +574,8 @@ func startAsking(ctx context.Context, open *stores, server *link.Server, conn *n
fmt.Printf("the operator's answers cannot be heard, so nothing is asked: %v\n", err)
return
}
askerFrom = a
go a.keep(ctx)
askerFrom.Store(a)
goServing(ctx, func() { a.keep(ctx) })
}
// servingAsker is the serving controller's asker, wired to the mesh: what it reads, publishes, performs and
+6 -2
View File
@@ -49,7 +49,7 @@ func attachTheChain(ctx context.Context, inv *inventory.Inventory, server *link.
// keys laid out before skips them, so this takes seconds; one that fails is tried again beside the serving, loudly.
if err := inv.LayBuildsOut(ctx, func(format string, args ...any) { fmt.Printf("chain: "+format+"\n", args...) }); err != nil {
fmt.Printf("chain: the builds are not laid out under their modules yet, and none is read until they are: %v\n", err)
go layBuildsOutSoon(ctx, inv)
goServing(ctx, func() { layBuildsOutSoon(ctx, inv) })
}
// The requests the move of delivery 1 found never handed over are not asked again: the controller asks no build
// (novox/hq ADR 0501; issue 1851, A10), and the builder builds every fingerprint the indexer writes.
@@ -80,7 +80,11 @@ func layBuildsOutSoon(ctx context.Context, inv *inventory.Inventory) {
Source: "chain", Summary: "the builds are not laid out under their modules, so none is read: "}
if err != nil {
fmt.Printf("chain: the builds are not laid out under their modules, and none is read until they are: %v\n", err)
if keeper := conditionsFrom; keeper != nil && !failed {
if ctx.Err() != nil {
return // stopped: the next controller lays them out
}
// A cancelled layout is tried again and raises nothing: a cancellation is no fault of the chain (issue 1945).
if keeper := conditionsFrom; keeper != nil && !failed && !errors.Is(err, context.Canceled) {
o.Summary += err.Error()
if _, oerr := keeper.Observe(ctx, o); oerr == nil {
failed = true
+8 -3
View File
@@ -194,7 +194,12 @@ func followDoneKey(ctx context.Context, open *stores, module string) {
}
continue
}
if keeper := conditionsFrom; keeper != nil && !failed {
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
}
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
@@ -308,10 +313,10 @@ func watchDoneKeys(ctx context.Context, open *stores) {
fmt.Printf("done keys: the builder's keys are not watched: %v\n", err)
return
}
go func() {
goServing(ctx, func() {
<-ctx.Done()
closeKeys()
}()
})
// And every module once now, after the watch is open so no write falls between the two.
inBackground(func() {
doneKeysAtStart(ctx, catalogedModules(open), func(module string) { takeDoneKeyAtStart(ctx, open, module) })
@@ -0,0 +1,61 @@
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)
}
})
}
}
+1 -1
View File
@@ -99,7 +99,7 @@ func TestAChangedUserListIsANewRevisionOfTheBusModulesCompositionAlone(t *testin
d["resources"] = append(d["resources"].([]any), map[string]any{"id": "nats.bus-users", "type": "file",
"module": "nats", "path": "/var/lib/mesh/nats/users.conf", "content": users, "mode": "0600"})
raw, _ := json.Marshal(d)
comps, _, err := compositionsOf("g14", raw, fps, nil, nil)
comps, _, _, err := compositionsOf("g14", raw, fps, nil, nil, nil)
if err != nil {
t.Fatal(err)
}
+32 -1
View File
@@ -119,7 +119,7 @@ func composeModulesByNode(ctx context.Context, open *stores, nodes []string) (ma
if err := open.inventory.RecordComposedPlans(ctx, node, plansKept(declared)); err != nil {
errsOf[node] = append(errsOf[node], fmt.Errorf("%s: its provisioner's plans: %w", node, err))
}
comps, moved, err := compositionsOf(node, body, fps, manifests, declared.modules)
comps, moved, left, err := compositionsOf(node, body, fps, manifests, declared.modules, namedByOwnComposition(ctx, at))
if err != nil {
errsOf[node] = append(errsOf[node], fmt.Errorf("%s: %w", node, err))
return
@@ -128,6 +128,11 @@ func composeModulesByNode(ctx context.Context, open *stores, nodes []string) (ma
fmt.Printf("compositions: %s's %s compose with the node's own: no fingerprint of the new form names them\n",
node, strings.Join(moved, ", "))
}
if len(left) > 0 {
fmt.Printf("compositions: %s's %s are left out of the node's own: the controller knows no fingerprint of the "+
"new form for them, and the node's want or have names their own composition (issue 1945)\n",
node, strings.Join(left, ", "))
}
if composeTrace != nil {
traceComposed(ctx, at, node, comps)
}
@@ -156,6 +161,32 @@ func composeModulesByNode(ctx context.Context, open *stores, nodes []string) (ma
return out, byNode
}
// namedByOwnComposition says a module's want or have on the node names a composition of its own: a fingerprint of the
// new form (issue 1945).
func namedByOwnComposition(ctx context.Context, at composedPlace) func(module string) (bool, error) {
return func(module string) (bool, error) {
if at.wanted != nil {
fp, err := at.wanted(ctx, module)
if err != nil {
return false, err
}
if strings.HasPrefix(fp, sdkdelivery.SourcePrefix2) {
return true, nil
}
}
if at.ran != nil {
h, found, err := at.ran(ctx, module)
if err != nil {
return false, err
}
if found && strings.HasPrefix(h.FP, sdkdelivery.SourcePrefix2) {
return true, nil
}
}
return false, nil
}
}
// composeAndSay is composeModules for an act that has nothing to do with its answer but say it: what each former send
// of a verb or an event calls.
func composeAndSay(ctx context.Context, open *stores, nodes []string) error {
+3 -3
View File
@@ -26,7 +26,7 @@ func TestALargeFileOfACompositionHasItsObjectInTheBucket(t *testing.T) {
if err != nil {
t.Fatal(err)
}
comps, _, err := compositionsOf("novox", body, nil, nil, nil)
comps, _, _, err := compositionsOf("novox", body, nil, nil, nil, nil)
if err != nil {
t.Fatal(err)
}
@@ -50,7 +50,7 @@ func TestACompositionNamingAnObjectTheBucketLacksIsRefused(t *testing.T) {
if err != nil {
t.Fatal(err)
}
comps, _, err := compositionsOf("novox", body, nil, nil, nil)
comps, _, _, err := compositionsOf("novox", body, nil, nil, nil, nil)
if err != nil {
t.Fatal(err)
}
@@ -78,7 +78,7 @@ func TestEveryFilledFileStaysFilledInTheCompositions(t *testing.T) {
for _, m := range []string{"power", "claude-code", "node-env"} {
delete(fps, m)
}
comps, _, err := compositionsOf(node, body, fps, nil, nil)
comps, _, _, err := compositionsOf(node, body, fps, nil, nil, nil)
if err != nil {
t.Fatalf("%s: %v", node, err)
}
+2 -2
View File
@@ -134,7 +134,7 @@ func watchHavesForJudgement(ctx context.Context, open *stores) {
fmt.Printf("compositions: the haves are not watched: %v\n", err)
return
}
go func() {
goServing(ctx, func() {
defer bus.Close()
defer w.Stop()
for {
@@ -165,5 +165,5 @@ func watchHavesForJudgement(ctx context.Context, open *stores) {
})
}
}
}()
})
}
@@ -608,8 +608,11 @@ func keepWhatServeSets(t *testing.T) {
t.Setenv(servedVar, "")
unheld, status, given, check, follow, keeper, hand := logUnheldChanges, statusFrom, givenEvents, checkEvents,
link.Calls.Follow, conditionsFrom, handActConn
doctor, asker := doctorFrom, askerFrom.Load()
t.Cleanup(func() {
logUnheldChanges, statusFrom, givenEvents, checkEvents, link.Calls.Follow, conditionsFrom, handActConn =
unheld, status, given, check, follow, keeper, hand
doctorFrom = doctor
askerFrom.Store(asker)
})
}
+40 -14
View File
@@ -67,8 +67,15 @@ type composedPlace struct {
// fingerprint fps gives it, the node's own with its adoption and provisioner, and every resource of a module fps names
// no fingerprint for in the node's own, its module taken off it and named in moved. Data says the module's manifest
// declares data.
func compositionsOf(node string, body []byte, fps map[string]string, manifests map[string]catalogue.Manifest, runs []string) (
out map[string]sdkdelivery.Composed, moved []string, err error) {
//
// **A module a want or a have names by its own composition is never in the node's own** (issue 1945): the node-engine
// runs that module's composition, and the same resources in the node's own were declared twice, which it refuses
// whole ("2 resources declare X", every node's apply failing from 2026-10-25 12:43). named says, for a module fps names
// no fingerprint of the new form for, that the node's want or have names a fingerprint of the new form; such a module's
// resources and filled paths are left out of this pass (its own composition, as written before, stays what the node
// runs) and it is named in left.
func compositionsOf(node string, body []byte, fps map[string]string, manifests map[string]catalogue.Manifest, runs []string,
named func(module string) (bool, error)) (out map[string]sdkdelivery.Composed, moved, left []string, err error) {
var envelope struct {
Resources []map[string]json.RawMessage `json:"resources"`
@@ -77,9 +84,9 @@ func compositionsOf(node string, body []byte, fps map[string]string, manifests m
Provisioner string `json:"provisioner_account"`
}
if err := json.Unmarshal(body, &envelope); err != nil {
return nil, nil, fmt.Errorf("%s's declaration: %w", node, err)
return nil, nil, nil, fmt.Errorf("%s's declaration: %w", node, err)
}
inModule := map[string]bool{}
inModule, ownComposition := map[string]bool{}, map[string]bool{}
resources := make([]json.RawMessage, 0, len(envelope.Resources))
for _, r := range envelope.Resources {
var module string
@@ -87,15 +94,30 @@ func compositionsOf(node string, body []byte, fps map[string]string, manifests m
_ = json.Unmarshal(raw, &module)
}
if module != "" && !strings.HasPrefix(fps[module], sdkdelivery.SourcePrefix2) {
delete(r, "module")
if !inModule[module] {
inModule[module] = true
moved = append(moved, module)
if !inModule[module] && !ownComposition[module] {
isNamed := false
if named != nil {
if isNamed, err = named(module); err != nil {
return nil, nil, nil, fmt.Errorf("whether %s's want or have names %s's own composition: %w",
node, module, err)
}
}
if isNamed {
ownComposition[module] = true
left = append(left, module)
} else {
inModule[module] = true
moved = append(moved, module)
}
}
if ownComposition[module] {
continue
}
delete(r, "module")
}
raw, err := json.Marshal(r)
if err != nil {
return nil, nil, err
return nil, nil, nil, err
}
resources = append(resources, raw)
}
@@ -106,22 +128,26 @@ func compositionsOf(node string, body []byte, fps map[string]string, manifests m
// and shanks).
var filled, ownFilled []string
for _, p := range envelope.Filled {
if filledByMoved(envelope.Resources, p, inModule) {
switch {
case filledByMoved(envelope.Resources, p, ownComposition):
// The file of a module whose own composition the node runs: that composition names it, not this pass.
case filledByMoved(envelope.Resources, p, inModule):
ownFilled = append(ownFilled, p)
} else {
default:
filled = append(filled, p)
}
}
sort.Strings(moved)
sort.Strings(left)
out, err = sdkdelivery.SplitComposed(node, resources, fps, filled,
sdkdelivery.NodeOwn{Adoption: envelope.Adoption, Provisioner: envelope.Provisioner})
if err != nil {
return nil, nil, err
return nil, nil, nil, err
}
if len(ownFilled) > 0 {
own, err := sdkdelivery.ComposedNodeKey(node)
if err != nil {
return nil, nil, err
return nil, nil, nil, err
}
c := out[own]
c.Filled = append(c.Filled, ownFilled...)
@@ -149,7 +175,7 @@ func compositionsOf(node string, body []byte, fps map[string]string, manifests m
out[key] = c
}
}
return out, moved, nil
return out, moved, left, nil
}
// filledByMoved says a filled path is a file of a module whose resources were moved into the node's own composition.
+54 -3
View File
@@ -218,7 +218,7 @@ func TestTheCompositionsOfANodeAreItsDeclaration(t *testing.T) {
Resources []json.RawMessage `json:"resources"`
}
_ = json.Unmarshal(body, &d)
comps, moved, err := compositionsOf(node, body, fps, nil, nil)
comps, moved, _, err := compositionsOf(node, body, fps, nil, nil, nil)
if err != nil {
t.Fatalf("%s: %v", node, err)
}
@@ -253,7 +253,7 @@ func TestTheCompositionsOfANodeAreItsDeclaration(t *testing.T) {
// unchanged it writes nothing; a changed input writes a new revision of that key alone, by compare-and-set.
func TestACompositionIsWrittenOnceAndAgainOnlyWhenItChanges(t *testing.T) {
body, fps := capturedBody(t, "g14")
comps, _, err := compositionsOf("g14", body, fps, nil, nil)
comps, _, _, err := compositionsOf("g14", body, fps, nil, nil, nil)
if err != nil {
t.Fatal(err)
}
@@ -342,7 +342,7 @@ func TestCompositionsOutsideTheFiveNewestAreDeleted(t *testing.T) {
func TestAModuleWithNoResourceOfItsOwnStillHasItsComposition(t *testing.T) {
body := []byte(`{"declaration":1,"resources":[]}`)
fp := sdkdelivery.SourcePrefix2 + strings.Repeat("b", 64)
comps, _, err := compositionsOf("laptop", body, map[string]string{"board": fp}, nil, []string{"board", "legacy"})
comps, _, _, err := compositionsOf("laptop", body, map[string]string{"board": fp}, nil, []string{"board", "legacy"}, nil)
if err != nil {
t.Fatal(err)
}
@@ -360,3 +360,54 @@ func TestAModuleWithNoResourceOfItsOwnStillHasItsComposition(t *testing.T) {
}
}
}
// 1945/AC16 (unit): a module the controller knows no fingerprint of the new form for, whose want or have on the node
// names its own composition, is never in the node's own composition, neither its resources nor its filled paths, and is
// named as left; a module neither names still goes with the node's own, its filled path with it. (Live from 2026-10-25
// 12:43: zsh, power, restic and more were in both, and every node refused "2 resources declare X".)
func TestAModuleItsWantNamesIsNeverInTheNodesOwnComposition(t *testing.T) {
body := []byte(`{"declaration":1,"resources":[
{"id":"zsh.rc","type":"file","module":"zsh","path":"/etc/zsh/zshrc"},
{"id":"power.unit","type":"file","module":"power","path":"/etc/power.conf"},
{"id":"mesh-wireguard.conf","type":"file","module":"mesh-wireguard","path":"/etc/wireguard/mesh0.conf"}],
"filled_files":["/etc/zsh/zshrc","/etc/power.conf","/etc/wireguard/mesh0.conf"]}`)
asked := map[string]int{}
named := func(module string) (bool, error) {
asked[module]++
return module == "zsh" || module == "power", nil
}
comps, moved, left, err := compositionsOf("ace", body, map[string]string{}, nil, nil, named)
if err != nil {
t.Fatal(err)
}
if strings.Join(left, ",") != "power,zsh" || strings.Join(moved, ",") != "mesh-wireguard" {
t.Errorf("left %v, moved %v; want power,zsh left and mesh-wireguard moved", left, moved)
}
if asked["zsh"] != 1 || asked["power"] != 1 {
t.Errorf("asked each module once: %v", asked)
}
own, _ := sdkdelivery.ComposedNodeKey("ace")
for key, c := range comps {
for _, id := range resourceIDs(t, c.Resources) {
if strings.HasPrefix(id, "zsh.") || strings.HasPrefix(id, "power.") {
t.Errorf("%s holds %s, which the module's own composition declares", key, id)
}
}
for _, p := range c.Filled {
if p == "/etc/zsh/zshrc" || p == "/etc/power.conf" {
t.Errorf("%s names %s as filled, which the module's own composition names", key, p)
}
}
}
if ids := strings.Join(resourceIDs(t, comps[own].Resources), " "); ids != "mesh-wireguard.conf" {
t.Errorf("the node's own composition holds %q, want mesh-wireguard.conf alone", ids)
}
if f := strings.Join(comps[own].Filled, " "); f != "/etc/wireguard/mesh0.conf" {
t.Errorf("the node's own composition names %q as filled, want mesh-wireguard's file alone", f)
}
if _, _, _, err := compositionsOf("ace", body, map[string]string{}, nil, nil,
func(string) (bool, error) { return false, fmt.Errorf("the wants bucket is gone") }); err == nil {
t.Errorf("a want that cannot be read composed the node anyway")
}
}
+1 -1
View File
@@ -49,7 +49,7 @@ func keeperOn(ctx context.Context, conn *nats.Conn, looks conditions.LookStore)
Say: func(format string, args ...any) { fmt.Fprintf(os.Stderr, format+"\n", args...) },
// What status leads with changed: composed again soon (a nudge outside the serving controller
// does nothing).
Changed: func() { statusFrom.nudge(); askerFrom.nudge() },
Changed: func() { statusFrom.nudge(); askerFrom.Load().nudge() },
// Written under the lease, carrying its epoch (novox/hq to-be 45 §6).
Epoch: func() (uint64, error) { return theLease.epoch(context.WithoutCancel(ctx)) }}), nil
}
+3
View File
@@ -225,6 +225,9 @@ func filesNotPutObservation(node, summary string) conditions.Observation {
// notePut records how putting a send's objects went: a failure counts, the second in a row raises the condition, a success
// clears the count and the condition.
func notePut(ctx context.Context, node string, put error) {
if errors.Is(put, context.Canceled) || ctx.Err() != nil {
return // a send cut by a cancellation or a stop says nothing of the bucket (issue 1945)
}
filesFailMu.Lock()
if put == nil {
delete(filesFailed, node)
@@ -415,6 +415,7 @@ func TestAPutThatFailsSendAfterSendRaisesAConditionAndClearsWhenItWorks(t *testi
was := conditionsFrom
conditionsFrom = k
t.Cleanup(func() { conditionsFrom = was })
forgetPutFailures(t, "cond-node")
key := filesNotPutObservation("cond-node", "").Key()
open := func() bool {
_, found, err := conditions.ReadOne(t.Context(), store, key)
@@ -442,6 +443,36 @@ func TestAPutThatFailsSendAfterSendRaisesAConditionAndClearsWhenItWorks(t *testi
}
}
// A send cut by a cancellation counts as no failure to put its files, however often (issue 1945: a cancellation is no fault
// of the bucket).
func TestAPutCutByACancellationRaisesNothing(t *testing.T) {
store := conditions.NewInMemory()
k := conditions.NewKeeper(t.Context(), conditions.Options{Store: store, History: store, Teller: &conditions.Told{}})
t.Cleanup(func() { k.Close(context.Background()) })
was := conditionsFrom
conditionsFrom = k
t.Cleanup(func() { conditionsFrom = was })
forgetPutFailures(t, "cut-node")
cut := fmt.Errorf("the files could not be put: %w", context.Canceled)
for range filesNotPutAfter + 1 {
notePut(t.Context(), "cut-node", cut)
}
if _, found, err := conditions.ReadOne(t.Context(), store, filesNotPutObservation("cut-node", "").Key()); err != nil || found {
t.Fatalf("sends cut by a cancellation raised the condition (found %v, %v)", found, err)
}
}
// forgetPutFailures starts the node's count of failed puts at nothing and leaves it so, as a run of the test again finds it.
func forgetPutFailures(t *testing.T, node string) {
forget := func() {
filesFailMu.Lock()
delete(filesFailed, node)
filesFailMu.Unlock()
}
forget()
t.Cleanup(forget)
}
// aBusThatSays is a bus that keeps the declarations published to it and, at each, asks whether the node's bucket is held.
type aBusThatSays struct {
link.Bus
@@ -87,3 +87,29 @@ func TestTheDeclaredPackagesMirrorNamesWhatEachNodesModulesDeclare(t *testing.T)
}
}
}
// 1945/AC-mirror-5: the digest the declared-packages mirror skips an unchanged pass by is made from a real reading of the
// stores, is the same over two readings of the same records, and moves when an assignment does.
func TestTheInputDigestOfTheStoresIsStable(t *testing.T) {
open := aMesh(t)
ctx := t.Context()
register(t, open, catalogue.Manifest{Module: "viewer", Version: "1", Resources: []map[string]any{
{"id": "pkg-curl", "type": "package", "package": "curl"}}})
read := func() string {
in, err := readComposeInput(ctx, open)
if err != nil {
t.Fatal(err)
}
return inputDigest(in)
}
first, second := read(), read()
if first == "" || first != second {
t.Fatalf("two readings of the same records gave the digests %q and %q", first, second)
}
if _, err := assign(ctx, open, "anchor", "viewer"); err != nil {
t.Fatalf("assign: %v", err)
}
if read() == first {
t.Errorf("an assignment left the digest of the stores' Input at %s", first)
}
}
@@ -0,0 +1,93 @@
package main
import (
"context"
"errors"
"fmt"
"testing"
sdkdelivery "git.novox.be/novox/mesh-sdk/go/delivery"
"github.com/novox/mesh-controller/internal/builderkeys"
)
// 1945/AC15 (unit): a module registered at a commit no build record points at (a catalogue merge re-indexed it with
// the fingerprint it had, and the take-in registered it from the done key whose record stands at its first commit)
// has, as its current fingerprint, the one the indexer's trunk value at that commit names; a module whose commit has a
// build keeps that build's; a module the indexer names nothing of the new form for has none; an indexer error is the
// answer's, never a module silently left out.
func TestACommitWithNoBuildOfItsOwnTakesTheIndexersFingerprint(t *testing.T) {
zsh, board := grantFP('a'), grantFP('b')
commits := map[string]string{"zsh": "68f7431d98d7", "board": "2167b95a1111", "legacy": "abcdef012345", "held": ""}
built := func(_ context.Context, module, commit string) (string, bool, error) {
if module == "board" {
return board, true, nil
}
return "", false, nil
}
indexed := func(_ context.Context, module, commit string) (string, error) {
switch module {
case "zsh":
return zsh, nil
case "board":
t.Errorf("the indexer was asked for board, whose commit has a build")
case "held":
t.Errorf("the indexer was asked for a module with no commit")
}
return "", nil
}
got, err := fingerprintsAt(context.Background(), commits, built, indexed)
if err != nil {
t.Fatal(err)
}
if got["zsh"] != zsh || got["board"] != board || len(got) != 2 {
t.Errorf("got %v, want zsh at %s and board at %s alone", got, zsh, board)
}
broken := func(context.Context, string, string) (string, error) { return "", errors.New("the bus is gone") }
if _, err := fingerprintsAt(context.Background(), commits, built, broken); err == nil {
t.Errorf("an indexer error was not the answer's")
}
}
// fakeTrunk is the indexer's trunk values of one module: now, and its history by commit.
type fakeTrunk struct {
now *sdkdelivery.Module
history map[string]sdkdelivery.Module
asked []string
}
func (f *fakeTrunk) ModuleAt(_ context.Context, module, commit string) (sdkdelivery.Module, error) {
f.asked = append(f.asked, commit)
if commit == "" {
if f.now == nil {
return sdkdelivery.Module{}, fmt.Errorf("%w (%s)", builderkeys.ErrNoValue, module)
}
return *f.now, nil
}
m, ok := f.history[commit]
if !ok {
return sdkdelivery.Module{}, fmt.Errorf("%w (%s at %s)", builderkeys.ErrNoValue, module, commit)
}
return m, nil
}
// 1945/AC15 (unit): the indexer's fingerprint at a commit is read from the value it holds now when that is the
// commit's (one read), else from the key's history at the commit; none held is no fingerprint.
func TestTheIndexersFingerprintAtACommit(t *testing.T) {
now := sdkdelivery.Module{Commit: "68f7431d98d7a1279cba3a615b6e9f22c3295783", FP2: grantFP('a')}
f := &fakeTrunk{now: &now}
if fp, err := indexedFingerprintAt(context.Background(), f, "zsh", "68f7431d98d7a1279cba3a615b6e9f22c3295783"); err != nil || fp != grantFP('a') || len(f.asked) != 1 {
t.Errorf("now's commit: %q (%v), asked %v", fp, err, f.asked)
}
f = &fakeTrunk{now: &now, history: map[string]sdkdelivery.Module{"2167b95a": {Commit: "2167b95a", FP2: grantFP('c')}}}
if fp, err := indexedFingerprintAt(context.Background(), f, "zsh", "2167b95a"); err != nil || fp != grantFP('c') {
t.Errorf("an older commit: %q (%v)", fp, err)
}
if fp, err := indexedFingerprintAt(context.Background(), f, "zsh", "99999999"); err != nil || fp != "" {
t.Errorf("a commit the indexer never wrote: %q (%v)", fp, err)
}
if fp, err := indexedFingerprintAt(context.Background(), &fakeTrunk{}, "zsh", "68f7431d"); err != nil || fp != "" {
t.Errorf("no value at all: %q (%v)", fp, err)
}
}
+2 -2
View File
@@ -169,7 +169,7 @@ func watchHavesForGiven(ctx context.Context, open *stores) {
log.Printf("given values: the haves are not watched: %v", err)
return
}
go func() {
goServing(ctx, func() {
defer bus.Close()
defer w.Stop()
for {
@@ -198,5 +198,5 @@ func watchHavesForGiven(ctx context.Context, open *stores) {
inBackground(func() { replaceGiven(ctx, open, node, map[string]time.Time{module: at}) })
}
}
}()
})
}
+108 -8
View File
@@ -11,6 +11,7 @@ import (
"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/inventory"
)
@@ -55,29 +56,128 @@ func grantsFrom(principals []broker.Principal, hashes map[string]string, current
// currentFingerprints is, by module, the fingerprint of the build each registered module is registered at: its
// manifest's commit (built_from), that commit's newest successful build, and the fingerprint that build names.
//
// **A commit with no build of its own asks the indexer** (issue 1945, live from 2026-10-25 12:43). A catalogue merge
// that leaves a module's sources alone re-indexes it at the merge's commit with the fingerprint it had: the take-in
// registers the module at the new commit from the same done key, whose build record (one per key and revision) already
// stands at the commit it was first taken in at, so the new commit has no build pointing at it. Read only by the
// commit, such a module had no fingerprint of the new form: it was composed into the node's own composition while its
// want named its own, and every node declared its files twice. The indexer's trunk value at the commit names the
// fingerprint the module was registered from, so that is its fingerprint.
func currentFingerprints(ctx context.Context, inv *inventory.Inventory) (map[string]string, error) {
builds, err := inv.CurrentBuilds(ctx)
if err != nil {
return nil, err
}
out := map[string]string{}
commits := make(map[string]string, len(builds))
for module, b := range builds {
if b.Commit == "" {
commits[module] = b.Commit
}
var indexed func(ctx context.Context, module, commit string) (string, error)
var closeIndexed func()
defer func() {
if closeIndexed != nil {
closeIndexed()
}
}()
return fingerprintsAt(ctx, commits,
func(ctx context.Context, module, commit string) (string, bool, error) {
id, err := inv.BuildOf(ctx, module, commit)
if err != nil || id == "" {
return "", false, nil
}
build, found, err := inv.BuildByID(ctx, id)
if err != nil || !found {
return "", false, nil
}
return build.SourceFingerprint, true, nil
},
func(ctx context.Context, module, commit string) (string, error) {
if indexed == nil {
read, done, err := openIndexedFingerprints()
if err != nil {
return "", err
}
indexed, closeIndexed = read, done
}
return indexed(ctx, module, commit)
})
}
// fingerprintsAt is, by module, the fingerprint of the build registered at its commit (built), or else the fingerprint
// of the new form the indexer's trunk value at that commit names (indexed). A module with no commit (held without a
// source) has none, and a module neither knows has none. An error of the indexer's is the answer's: a module left out
// for it would be composed into the node's own composition beside its own (issue 1945).
func fingerprintsAt(ctx context.Context, commits map[string]string,
built func(ctx context.Context, module, commit string) (string, bool, error),
indexed func(ctx context.Context, module, commit string) (string, error)) (map[string]string, error) {
modules := make([]string, 0, len(commits))
for module := range commits {
modules = append(modules, module)
}
sort.Strings(modules)
out := map[string]string{}
for _, module := range modules {
commit := commits[module]
if commit == "" {
continue
}
id, err := inv.BuildOf(ctx, module, b.Commit)
if err != nil || id == "" {
fp, found, err := built(ctx, module, commit)
if err != nil {
return nil, err
}
if found {
out[module] = fp
continue
}
build, found, err := inv.BuildByID(ctx, id)
if err != nil || !found {
continue
fp, err = indexed(ctx, module, commit)
if err != nil {
return nil, fmt.Errorf("the fingerprint %s is registered at (%.8s): %w", module, commit, err)
}
if strings.HasPrefix(fp, sdkdelivery.SourcePrefix2) {
out[module] = fp
}
out[module] = build.SourceFingerprint
}
return out, nil
}
// openIndexedFingerprints opens the indexer's trunk values: what reads the fingerprint of the new form a module's value
// at a commit names ("" when it holds none for that commit), and what closes the bus connection it opened.
var openIndexedFingerprints = func() (func(ctx context.Context, module, commit string) (string, error), func(), error) {
keys, closeKeys, err := openBuilderKeys()
if err != nil {
return nil, nil, fmt.Errorf("the indexer's trunk values: %w", err)
}
return func(ctx context.Context, module, commit string) (string, error) {
return indexedFingerprintAt(ctx, keys, module, commit)
}, closeKeys, nil
}
// indexedFingerprintAt is the fingerprint of the new form the indexer's value of module at commit names: the value it
// holds now when that is the commit's (one read), else the newest of the key's history at the commit; "" when none.
func indexedFingerprintAt(ctx context.Context, keys interface {
ModuleAt(ctx context.Context, module, commit string) (sdkdelivery.Module, error)
}, module, commit string) (string, error) {
m, err := keys.ModuleAt(ctx, module, "")
if err == nil && sameCommitID(m.Commit, commit) {
return m.FP2, nil
}
if err != nil && !errors.Is(err, builderkeys.ErrNoValue) {
return "", err
}
if errors.Is(err, builderkeys.ErrNoValue) {
return "", nil
}
m, err = keys.ModuleAt(ctx, module, commit)
if errors.Is(err, builderkeys.ErrNoValue) {
return "", nil
}
if err != nil {
return "", err
}
return m.FP2, nil
}
// grantsKV is what writing the grants needs of their bucket.
type grantsKV interface {
Get(ctx context.Context, key string) (jetstream.KeyValueEntry, error)
+2 -2
View File
@@ -223,10 +223,10 @@ func watchNewModules(ctx context.Context, open *stores) {
fmt.Printf("new modules: the builder's keys are not watched: %v\n", err)
return
}
go func() {
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
+11
View File
@@ -59,6 +59,17 @@ func dataWords(o conditions.Observation) (what, module string) {
}
var plainWordings = map[string]func(conditions.Observation) words{
// A module's new build whose compositions or grant the controller could not write (composed_keys.go).
kindNotComposed: worded(func(o conditions.Observation) words {
m := "a module"
if p := idPart(o, 0); o.Scope == conditions.ScopeBuild && p != "" {
m = p // the module: raised in the build's scope, its id the module's name
}
return words{Headline: "A new build of " + m + " cannot be delivered",
Explanation: fmt.Sprintf("A new build of %s was made, and the mesh could not prepare it for the machines, so "+
"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"}
}),
// Machines and their node-engine.
"silent": worded(func(o conditions.Observation) words {
m := machineOr(o, "a machine")
+31 -14
View File
@@ -122,7 +122,15 @@ func serve(ctx context.Context) (err error) {
defer theLease.release()
ctx, stopActing := context.WithCancel(ctx)
defer stopActing()
go func() {
// Every goroutine started from here is counted, and ended and waited for before serve returns (serving_group.go):
// on the way out by error here, and at the end before anything else is undone.
ctx, running := withServingGroup(ctx)
stopRunning := func() {
stopActing()
running.stop()
}
defer stopRunning()
goServing(ctx, func() {
select {
case <-lost:
stopActing()
@@ -132,7 +140,7 @@ func serve(ctx context.Context) (err error) {
// asks the controller after this one rather than have a command started and killed with this
// process (novox/hq issue 289).
handingOver.Store(true)
}()
})
defer func() {
select {
case <-lost:
@@ -159,6 +167,16 @@ func serve(ctx context.Context) (err error) {
if err := attachTheChain(ctx, inv, server); err != nil {
return err
}
bus, isNATS := server.Bus().(link.OverNATS)
if !isNATS {
return errors.New("the mesh's verbs are served over the bus, and this control plane is not on it")
}
// The condition store, opened before anything that raises or reads a condition starts beside the serving, so none
// of them meets conditionsFrom being written (issue 1945); closed after they all ended.
keeper := openServingConditions(ctx, open, bus)
if keeper != nil {
defer closeConditions(keeper)
}
// The bus's own objects, asserted on every start. **Not created once at genesis**: a stream
// somebody deleted, a mesh raised from a restored backup, or a bus whose data directory was
@@ -173,7 +191,7 @@ func serve(ctx context.Context) (err error) {
// And the pending assignments settled on a tick of their own (novox/hq ADR 0261): made once their module
// is registered, ended with why when its build will not register it, raised and cleared as conditions.
// Never by a read.
go settlingPending(ctx, open)
goServing(ctx, func() { settlingPending(ctx, open) })
// The durations the core's bounds are set from are kept a month (novox/hq to-be 45 Phase 0).
// And what the catalogue decided a build meant. The builder's own result is already handled
// above; this is the other half — the control plane is the only one of the three that knows
@@ -221,10 +239,6 @@ func serve(ctx context.Context) (err error) {
"Those answer the reason when called; everything else is served as usual\n",
catalogue.ControllerSeatName, strings.Join(behind, ", "))
}
bus, isNATS := server.Bus().(link.OverNATS)
if !isNATS {
return errors.New("the mesh's verbs are served over the bus, and this control plane is not on it")
}
// The hand-act log is counted for `status` on this connection rather than a new one a minute; no longer serving, the
// connection is closed, and a verb run after in this process dials its own (a served test's leftover failed the
// next test's `conditions` with "connection closed", issue 1945).
@@ -244,18 +258,18 @@ func serve(ctx context.Context) (err error) {
return err
}
// Composed now and kept current, before the verb that answers from it is served.
go statusFrom.keep(ctx)
goServing(ctx, func() { statusFrom.keep(ctx) })
// The facts snapshot a merge check is fed (novox/hq to-be 45 §9): kept current while this
// controller holds the lease, read by the build seat from the artifact store.
go exportingFacts(ctx, open, bus.Conn.ConnectedServerVersion)
goServing(ctx, func() { exportingFacts(ctx, open, bus.Conn.ConnectedServerVersion) })
// The record mirrors (novox/hq ADR 0289, design 57 F3): the assignments and each registered module's
// declared config on the bus, written from the store only when a record changed, by the lease holder.
// The provisioner's plan of each handed module, as its machine was last sent it (design 57 P6), in the module's
// assignment.
inv.SetProvisioningPlans(inv.SentProvisioningPlans)
go mirroringRecords(ctx, open, inv, bus.Conn, bus)
goServing(ctx, func() { mirroringRecords(ctx, open, inv, bus.Conn, bus) })
// And the bucket of the files that travel by digest, judged against what each machine's latest declaration names (ADR 0486 §3).
go sweepingDeclarationFiles(ctx, inv, bus.Conn)
goServing(ctx, func() { sweepingDeclarationFiles(ctx, inv, bus.Conn) })
// And the modules the catalogue does not hold yet, registered from the builder's first done key of each, woken by
// the builder's and the indexer's writes (novox/hq ADR 0501 §6, issue 1851): nothing asks a first build.
watchNewModules(ctx, open)
@@ -287,7 +301,7 @@ func serve(ctx context.Context) (err error) {
// signals table, what the bus says about itself, and the self-check. A store that cannot be opened
// is said and the controller serves on: status then says the conditions cannot be read, and is
// not well — louder than not serving at all, and the push that repairs the bus still runs.
if stopWatching := watchTheMesh(ctx, open, server, bus); stopWatching != nil {
if stopWatching := watchTheMesh(ctx, open, server, bus, keeper); stopWatching != nil {
defer stopWatching()
}
stopServing, err := bus.ServeSeatTools(catalogue.ControllerSeatName, handlers, log.New(os.Stdout, "", log.LstdFlags))
@@ -302,7 +316,7 @@ func serve(ctx context.Context) (err error) {
}
defer stopCLI()
// Its read calls write no line each; how many each caller made is said once an hour (novox/hq issue 762).
go sayReadCalls(ctx)
goServing(ctx, func() { sayReadCalls(ctx) })
// And says so on the bus (novox/hq ADR 0197): what it serves, as the NATS services protocol asks.
stopAnnouncing, err := bus.Announce(seatAnnouncement(handlers), log.New(os.Stdout, "", log.LstdFlags))
if err != nil {
@@ -310,7 +324,10 @@ func serve(ctx context.Context) (err error) {
}
defer stopAnnouncing()
return server.Serve(ctx)
err = server.Serve(ctx)
// Ended and waited for before the verbs stop being served and the globals are put back, not after.
stopRunning()
return err
}
// OverlayCIDRVar is the range the mesh allocates node addresses from.
+266 -24
View File
@@ -2,12 +2,15 @@ package main
import (
"context"
"crypto/sha256"
"encoding/hex"
"encoding/json"
"fmt"
"sort"
"time"
"git.novox.be/novox/mesh-sdk/go/compose"
sdkdelivery "git.novox.be/novox/mesh-sdk/go/delivery"
"github.com/nats-io/nats.go"
"github.com/nats-io/nats.go/jetstream"
@@ -24,7 +27,8 @@ import (
// so no one call site sees every change; what every change has in common is that it is committed to the
// store. So the serving controller reads the store every mirrorEvery and writes only what differs from what
// the bucket holds (ADR 0289 §1): a quiet mesh writes nothing, and a change rolled back is never read. A
// mirror is therefore at most mirrorEvery behind its store.
// mirror is therefore at most mirrorEvery behind its store. The declared-packages mirror is the exception: it composes
// every node, so it passes on a change of what it composes from, not on this tick (keepingDeclaredPackages, issue 1945).
// mirrorEvery is how often the mirrors are compared with the store: two small reads of the store.
const mirrorEvery = 10 * time.Second
@@ -45,21 +49,31 @@ func mirroringRecords(ctx context.Context, open *stores, inv *inventory.Inventor
return e
}
sources := map[string]func(context.Context) (map[string][]byte, error){
recordmirror.DeclaredConfig.Name: inv.DeclaredConfigMirror,
recordmirror.Predefined.Name: inv.PredefinedMirror,
recordmirror.DeclaredPackages.Name: declaredPackagesMirror(open),
recordmirror.DeclaredConfig.Name: inv.DeclaredConfigMirror,
recordmirror.Predefined.Name: inv.PredefinedMirror,
}
packages := keepingDeclaredPackages(ctx, open, api, announce, epoch, holds)
held := false
writers := map[string]*recordmirror.Writer{}
failing := map[string]string{}
tick := time.NewTicker(mirrorEvery)
defer tick.Stop()
events := &assignmentEvents{}
for {
if holds() {
now := holds()
if now && !held {
// The lease taken: the declared-packages mirror is brought to the stores at once, not at its next change.
packages.poke()
}
held = now
if now {
// The assignments are keys of the bus, written by the verbs that change them (novox/hq ADR 0408): this tick
// keeps the provisioner's plan in their values and says what changed. They are no mirror of the store.
keepingAssignments(ctx, inv, announce, epoch, events, failing)
for _, m := range recordmirror.Mirrors() {
if sources[m.Name] == nil {
continue // the declared packages, kept on a change (keepingDeclaredPackages)
}
why := mirrorOnce(ctx, api, m, writers, sources[m.Name], announce, epoch, holds)
if why != failing[m.Name] {
if why != "" {
@@ -179,35 +193,263 @@ func keepingAssignments(ctx context.Context, inv *inventory.Inventory, announce
events.known = now
}
// declaredPackagesMirror is the source of the declared-packages mirror (ADR 0507 §5): each node's composition, asked as a
// question that makes nothing, and the packages it declares with the modules that declare them. A node whose own set does
// not compose has no key, and any other failure to read is the pass's error, so a key is never purged for a store that
// could not be read.
// declaredPackagesMirror is the source of the declared-packages mirror (ADR 0507 §5) over a fresh reading of the stores.
func declaredPackagesMirror(open *stores) func(context.Context) (map[string][]byte, error) {
return func(ctx context.Context) (map[string][]byte, error) {
in, err := inputFor(ctx, open)
if err != nil {
return nil, err
}
// Which module owns each resource is carried only for a node-engine that reads it, so every node is asked as one that does.
asked := in.Input
asked.Nodes = append([]compose.Node(nil), in.Nodes...)
for i := range asked.Nodes {
asked.Nodes[i].ReadsModules = true
}
out := map[string][]byte{}
for _, n := range in.sortedNodes() {
d, _, err := compose.ComposeIn(asked, n.Name, compose.Foreseeing)
if compose.IsUnresolvable(err) {
continue
return declaredPackagesFrom(ctx, in)
}
}
// declaredPackagesFrom is the declared-packages mirror over an Input: each node's composition, asked as a question that makes
// nothing, and the packages it declares with the modules that declare them. A node whose own set does not compose has no key,
// and any other failure to read is the pass's error, so a key is never purged for a store that could not be read.
func declaredPackagesFrom(_ context.Context, in *meshInput) (map[string][]byte, error) {
// Which module owns each resource is carried only for a node-engine that reads it, so every node is asked as one that does.
asked := in.Input
asked.Nodes = append([]compose.Node(nil), in.Nodes...)
for i := range asked.Nodes {
asked.Nodes[i].ReadsModules = true
}
out := map[string][]byte{}
for _, n := range in.sortedNodes() {
d, _, err := compose.ComposeIn(asked, n.Name, compose.Foreseeing)
if compose.IsUnresolvable(err) {
continue
}
if err != nil {
return nil, fmt.Errorf("the packages %s declares cannot be read: %w", n.Name, err)
}
if out[n.Name], err = inventory.DeclaredPackagesValue(n.Name, d.Resources, d.ModuleOf); err != nil {
return nil, err
}
}
return out, nil
}
// **The declared-packages mirror passes on a change, not on a tick** (issue 1945 note 22). Its source composes every node in
// full, about 45 s on the control node; on the 10 s tick it never rested and held the controller at 130-175 % CPU. Every
// change that can move a node's packages is a change of its composition, which the controller writes into the bucket of the
// compositions' digests, and of the assignments, which the verbs write into theirs: a watch of those two buckets wakes the
// keeper. A pass reads the stores' Input once and composes only when the Input's digest differs from the one it last
// mirrored, so a wake that changed nothing it composes from costs one reading. One pass at a time; the wakes during a pass
// are one pass after it, at least packagesGap after the last began. A rare safety pass (packagesSafetyEvery) covers a
// change no watched key says, and with an unchanged digest composes nothing either.
const (
packagesSafetyEvery = 10 * time.Minute
packagesGap = 30 * time.Second
)
// packagesTiming is when the declared-packages mirror passes: the safety pass's period and the least time between the
// starts of two passes.
type packagesTiming struct{ safety, gap time.Duration }
// packagesKeeper keeps the declared-packages mirror: read is the Input, compose the mirror's values from it, sync writes
// those that differ from the bucket, holds says this controller holds the lease. Each a seam a test stands in for.
type packagesKeeper struct {
read func(ctx context.Context) (*meshInput, error)
compose func(ctx context.Context, in *meshInput) (map[string][]byte, error)
sync func(ctx context.Context, want map[string][]byte) error
holds func() bool
wake chan struct{}
// last is the digest of the Input last mirrored whole; empty after a failure or while the lease is not held, so the
// next pass composes.
last string
}
func newPackagesKeeper(read func(context.Context) (*meshInput, error),
compose func(context.Context, *meshInput) (map[string][]byte, error),
sync func(context.Context, map[string][]byte) error, holds func() bool) *packagesKeeper {
return &packagesKeeper{read: read, compose: compose, sync: sync, holds: holds, wake: make(chan struct{}, 1)}
}
// poke says something the mirror composes from may have changed; never blocks, and pokes before a pass are one.
func (k *packagesKeeper) poke() {
select {
case k.wake <- struct{}{}:
default:
}
}
// pass brings the mirror to the Input, composing only when the Input differs from the one last mirrored.
func (k *packagesKeeper) pass(ctx context.Context) error {
if !k.holds() {
k.last = ""
return nil
}
in, err := k.read(ctx)
if err != nil {
k.last = ""
return fmt.Errorf("the store cannot be read: %w", err)
}
digest := inputDigest(in)
if digest != "" && digest == k.last {
return nil
}
want, err := k.compose(ctx, in)
if err != nil {
k.last = ""
return err
}
if err := k.sync(ctx, want); err != nil {
k.last = ""
return err
}
k.last = digest
return nil
}
// run passes once at its start, then on each wake and each safety period, until ctx ends. said hears why a pass failed,
// and an empty line when one succeeds again.
func (k *packagesKeeper) run(ctx context.Context, at packagesTiming, said func(why string)) {
safety := time.NewTicker(at.safety)
defer safety.Stop()
failing := ""
var began time.Time
for {
// The wakes so far are this pass's: drained before it reads, so a change after is a pass of its own.
select {
case <-k.wake:
default:
}
began = time.Now()
why := ""
if err := k.pass(ctx); err != nil {
why = err.Error()
}
if why != failing {
said(why)
failing = why
}
select {
case <-ctx.Done():
return
case <-k.wake:
case <-safety.C:
}
if rest := at.gap - time.Since(began); rest > 0 {
select {
case <-ctx.Done():
return
case <-time.After(rest):
}
}
}
}
// inputDigest is a digest of everything the composition reads (the Input, and why a node or the bus is refused): the same
// for the same Input. Empty when it cannot be made, and then the pass composes.
func inputDigest(in *meshInput) string {
refused := map[string]string{}
for node, err := range in.refused {
if err != nil {
refused[node] = err.Error()
}
}
bus := ""
if in.busRefused != nil {
bus = in.busRefused.Error()
}
body, err := json.Marshal(struct {
Input compose.Input
Refused map[string]string
BusRefused string
}{in.Input, refused, bus})
if err != nil {
return ""
}
sum := sha256.Sum256(body)
return hex.EncodeToString(sum[:])
}
// keepingDeclaredPackages keeps the declared-packages mirror for as long as ctx: woken by the watch of the compositions'
// digests and of the assignments. It answers the keeper, whose poke a caller may also wake it with.
func keepingDeclaredPackages(ctx context.Context, open *stores, api jetstream.JetStream, announce recordmirror.Announcer,
epoch func() uint64, holds func() bool) *packagesKeeper {
m := recordmirror.DeclaredPackages
var w *recordmirror.Writer
read := func(ctx context.Context) (*meshInput, error) {
reading, cancel := context.WithTimeout(ctx, 2*time.Minute)
defer cancel()
return readComposeInput(reading, open)
}
sync := func(ctx context.Context, want map[string][]byte) error {
writing, cancel := context.WithTimeout(ctx, 30*time.Second)
defer cancel()
if w == nil {
kv, err := api.KeyValue(writing, m.Bucket)
if err != nil {
return nil, fmt.Errorf("the packages %s declares cannot be read: %w", n.Name, err)
return fmt.Errorf("the bucket %s cannot be read: %v", m.Bucket, err)
}
w = &recordmirror.Writer{Mirror: m, KV: kv, Announce: announce, Epoch: epoch, Holds: holds}
}
lost := w.LostEvents
r, err := w.Sync(writing, want)
if r.Written+r.Removed > 0 {
fmt.Printf("the %s mirror moved: %d written, %d removed\n", m.Name, r.Written, r.Removed)
}
if w.LostEvents > lost {
fmt.Printf("the %s mirror lost %d event(s): the bucket holds the change, its notice was refused twice\n",
m.Name, w.LostEvents-lost)
}
return err
}
k := newPackagesKeeper(read, declaredPackagesFrom, sync, holds)
for _, bucket := range []string{sdkdelivery.ControllerComposedBucket, broker.AssignmentsBucket} {
goServing(ctx, func() { wakeOnBucket(ctx, api, bucket, k.poke) })
}
goServing(ctx, func() {
k.run(ctx, packagesTiming{safety: packagesSafetyEvery, gap: packagesGap}, func(why string) {
if why != "" {
fmt.Printf("the %s mirror cannot be kept: %s\n", m.Name, why)
} else {
fmt.Printf("the %s mirror is kept again\n", m.Name)
}
})
})
return k
}
// wakeOnBucket calls wake on every write or delete of a key in the bucket, for as long as ctx: only the keys' metadata is
// watched. A bucket not there yet, or a watch that ends, is watched again a minute later; the safety pass covers the gap.
func wakeOnBucket(ctx context.Context, api jetstream.JetStream, bucket string, wake func()) {
for {
if err := watchBucketOnce(ctx, api, bucket, wake); err != nil && ctx.Err() == nil {
fmt.Printf("the declared-packages mirror does not hear %s: %v\n", bucket, err)
}
select {
case <-ctx.Done():
return
case <-time.After(time.Minute):
}
}
}
// watchBucketOnce watches the bucket until its watch ends or ctx does.
func watchBucketOnce(ctx context.Context, api jetstream.JetStream, bucket string, wake func()) error {
kv, err := api.KeyValue(ctx, bucket)
if err != nil {
return err
}
watch, err := kv.WatchAll(ctx, jetstream.UpdatesOnly(), jetstream.MetaOnly())
if err != nil {
return err
}
defer func() { _ = watch.Stop() }()
for {
select {
case <-ctx.Done():
return nil
case e, ok := <-watch.Updates():
if !ok {
return fmt.Errorf("the watch ended")
}
if out[n.Name], err = inventory.DeclaredPackagesValue(n.Name, d.Resources, d.ModuleOf); err != nil {
return nil, err
if e != nil {
wake()
}
}
return out, nil
}
}
+191
View File
@@ -0,0 +1,191 @@
package main
import (
"context"
"fmt"
"sync"
"sync/atomic"
"testing"
"time"
"git.novox.be/novox/mesh-sdk/go/catalogue"
"git.novox.be/novox/mesh-sdk/go/compose"
)
// The declared-packages mirror composes only when what it composes from changed (issue 1945 note 22: it composed every node
// in full every 10 s, a pass of about 45 s, so the controller on the control node never rested at 130-175 % CPU).
// scaled is a cadence of the mirror with a second taken as a millisecond, so a test's minute is 60 ms.
func scaled() packagesTiming {
return packagesTiming{safety: packagesSafetyEvery / 1000, gap: packagesGap / 1000}
}
const scaledMinute = 60 * time.Millisecond
// fakeMesh is the keeper's input as a test changes it: each node's assigned modules, and the compositions counted by node.
type fakeMesh struct {
mu sync.Mutex
assigned map[string][]string
composed atomic.Int64 // node compositions
synced atomic.Int64
lastSync map[string][]byte
// block, when set, holds the next composition until it is closed; entered is closed when that composition began.
block chan struct{}
entered chan struct{}
}
func newFakeMesh(nodes int) *fakeMesh {
f := &fakeMesh{assigned: map[string][]string{}}
for i := 0; i < nodes; i++ {
f.assigned[fmt.Sprintf("node%d", i)] = []string{"base"}
}
return f
}
func (f *fakeMesh) assign(node, module string) {
f.mu.Lock()
defer f.mu.Unlock()
f.assigned[node] = append(append([]string(nil), f.assigned[node]...), module)
}
func (f *fakeMesh) keeper() *packagesKeeper {
read := func(context.Context) (*meshInput, error) {
f.mu.Lock()
defer f.mu.Unlock()
in := compose.Input{Catalogue: map[string]catalogue.Manifest{"base": {Module: "base", Version: "1"}}}
for _, n := range nodesOfFake(f.assigned) {
in.Nodes = append(in.Nodes, compose.Node{Name: n, Assigned: append([]string(nil), f.assigned[n]...)})
}
return &meshInput{Input: in}, nil
}
composeAll := func(_ context.Context, in *meshInput) (map[string][]byte, error) {
f.mu.Lock()
block, entered := f.block, f.entered
f.block = nil
f.mu.Unlock()
if block != nil {
close(entered)
<-block
}
out := map[string][]byte{}
for _, n := range in.Nodes {
f.composed.Add(1)
out[n.Name] = []byte(fmt.Sprint(n.Assigned))
}
return out, nil
}
syncAll := func(_ context.Context, want map[string][]byte) error {
f.mu.Lock()
f.lastSync = want
f.mu.Unlock()
f.synced.Add(1)
return nil
}
return newPackagesKeeper(read, composeAll, syncAll, func() bool { return true })
}
func nodesOfFake(m map[string][]string) []string {
out := make([]string, 0, len(m))
for k := range m {
out = append(out, k)
}
for i := range out {
for j := i + 1; j < len(out); j++ {
if out[j] < out[i] {
out[i], out[j] = out[j], out[i]
}
}
}
return out
}
func runKeeper(t *testing.T, k *packagesKeeper) context.CancelFunc {
ctx, cancel := context.WithCancel(context.Background())
done := make(chan struct{})
go func() {
defer close(done)
k.run(ctx, scaled(), func(string) {})
}()
t.Cleanup(func() {
cancel()
<-done
})
return cancel
}
func eventually(t *testing.T, what string, cond func() bool) {
t.Helper()
deadline := time.Now().Add(time.Second)
for !cond() {
if time.Now().After(deadline) {
t.Fatalf("not within a second: %s", what)
}
time.Sleep(time.Millisecond)
}
}
// 1945/AC-mirror-1: given a mesh whose input does not change, when a minute passes after the mirror's first pass, then it
// composes no node (before: a pass every 10 s, each composing every node). The count per minute is said.
func TestTheDeclaredPackagesMirrorComposesNothingWithoutAChange(t *testing.T) {
const nodes = 5
f := newFakeMesh(nodes)
runKeeper(t, f.keeper())
eventually(t, "the first pass", func() bool { return f.synced.Load() >= 1 })
start := f.composed.Load()
time.Sleep(scaledMinute)
perMinute := f.composed.Load() - start
t.Logf("node compositions in a minute of an unchanged mesh of %d nodes: %d", nodes, perMinute)
if perMinute != 0 {
t.Errorf("an unchanged mesh of %d nodes was composed %d times in a minute; want 0", nodes, perMinute)
}
}
// 1945/AC-mirror-2: given a pass that is running, when fifty changes arrive during it, then exactly one more pass follows,
// never two at once and never one per change.
func TestTheDeclaredPackagesMirrorPassesOncePerBurst(t *testing.T) {
f := newFakeMesh(1)
release := make(chan struct{})
f.block, f.entered = release, make(chan struct{})
k := f.keeper()
runKeeper(t, k)
<-f.entered
for i := 0; i < 50; i++ {
f.assign("node0", fmt.Sprintf("m%d", i))
k.poke()
}
close(release)
eventually(t, "the pass after the burst", func() bool { return f.synced.Load() >= 2 })
time.Sleep(2*scaled().gap + 20*time.Millisecond)
if got := f.synced.Load(); got != 2 {
t.Errorf("a burst of 50 changes during a pass gave %d passes in all; want 2 (the running one and one after)", got)
}
}
// 1945/AC-mirror-3: given the mirror at rest, when an input changes and its event arrives, then the change is mirrored.
func TestTheDeclaredPackagesMirrorMirrorsAChange(t *testing.T) {
f := newFakeMesh(2)
k := f.keeper()
runKeeper(t, k)
eventually(t, "the first pass", func() bool { return f.synced.Load() >= 1 })
f.assign("node1", "viewer")
k.poke()
eventually(t, "the change mirrored", func() bool {
f.mu.Lock()
defer f.mu.Unlock()
return string(f.lastSync["node1"]) == "[base viewer]"
})
}
// 1945/AC-mirror-4: the digest of the input is the same for the same input and differs when an assignment differs.
func TestTheInputDigestFollowsTheInput(t *testing.T) {
of := func(assigned ...string) *meshInput {
return &meshInput{Input: compose.Input{Nodes: []compose.Node{{Name: "anchor", Assigned: assigned}}}}
}
a, b, c := inputDigest(of("base")), inputDigest(of("base")), inputDigest(of("base", "viewer"))
if a == "" || a != b {
t.Errorf("the same input gave the digests %q and %q", a, b)
}
if a == c {
t.Errorf("an added assignment left the digest at %q", a)
}
}
+59
View File
@@ -0,0 +1,59 @@
package main
import (
"context"
"sync"
)
// **Every goroutine the serving controller starts ends before serve returns** (issue 1945). A loop or a watch started
// beside the serving runs until serve's context ends, and serve's stop waits for it before it closes the stores, the
// bus and the condition keeper those loops read: a loop that outlived its serve met the next serve's globals, its
// closed stores and its bus, and in the test process that is the next test (a race on conditionsFrom, a package that
// passed every test and still exited failing). Where it is checked: the tests that run serve whole, under -race.
// servingGroup counts the goroutines one serve started. Once its stop began it starts nothing more: what would start is
// work for a controller that is stopping.
type servingGroup struct {
mu sync.Mutex
stopped bool
wg sync.WaitGroup
}
type servingGroupKey struct{}
// withServingGroup is ctx carrying a new group, which goServing under it counts.
func withServingGroup(ctx context.Context) (context.Context, *servingGroup) {
g := &servingGroup{}
return context.WithValue(ctx, servingGroupKey{}, g), g
}
// goServing runs work beside the caller, counted in the group ctx carries, so serve's stop waits for it. Under a group
// whose stop began it runs nothing; under a context with no group (a part run alone, by a test or a verb) it is a plain
// goroutine, as before.
func goServing(ctx context.Context, work func()) {
g, _ := ctx.Value(servingGroupKey{}).(*servingGroup)
if g == nil {
go work()
return
}
g.mu.Lock()
if g.stopped {
g.mu.Unlock()
return
}
g.wg.Add(1)
g.mu.Unlock()
go func() {
defer g.wg.Done()
work()
}()
}
// stop starts nothing more and waits for what was started; the caller ends their context first. A second stop waits
// for nothing.
func (g *servingGroup) stop() {
g.mu.Lock()
g.stopped = true
g.mu.Unlock()
g.wg.Wait()
}
+2 -2
View File
@@ -73,10 +73,10 @@ func keepAnnouncements(ctx context.Context, js *broker.JetStream, open *stores)
if err != nil {
return fmt.Errorf("following the states announced: %w", err)
}
go func() {
goServing(ctx, func() {
<-ctx.Done()
consuming.Stop()
}()
})
return nil
}
+1 -1
View File
@@ -210,7 +210,7 @@ func (l nudgingListener) Heard(ctx context.Context, report link.Report) (bool, e
// And, once the move of the secrets into the vault has started, what is left of this machine's (design 57's V3,
// issue 676): a machine asleep when it started is moved when it is next heard from. Off the report's path.
if err == nil && l.open != nil && report.Superseded == "" && report.Rekey == nil && report.Node != "" {
go moveWhenHeard(context.WithoutCancel(ctx), l.open, moveHeardOn(), report.Node)
goServing(ctx, func() { moveWhenHeard(context.WithoutCancel(ctx), l.open, moveHeardOn(), report.Node) })
}
return news, err
}
+26 -11
View File
@@ -543,10 +543,9 @@ func later(a, b time.Time) time.Time {
return b
}
// watchTheMesh opens the condition store and starts the watchdogs, the bus's advisories and the
// self-check, for the serving controller; the returned function stops them. Nil when the store could
// not be opened, which is said.
func watchTheMesh(ctx context.Context, open *stores, server *link.Server, bus link.OverNATS) func() {
// openServingConditions opens the condition store for the serving controller and makes it conditionsFrom. Nil when the store
// could not be opened, which is said: status then says the conditions cannot be read.
func openServingConditions(ctx context.Context, open *stores, bus link.OverNATS) *conditions.Keeper {
keeper, err := keeperOn(ctx, bus.Conn, storeLooks{open.inventory})
if err != nil {
fmt.Printf("the condition store could NOT be opened, so nothing that goes wrong is kept or said, and "+
@@ -554,6 +553,22 @@ func watchTheMesh(ctx context.Context, open *stores, server *link.Server, bus li
return nil
}
conditionsFrom = keeper
return keeper
}
// closeConditions flushes what the keeper holds and closes it, once nothing that raises a condition runs.
func closeConditions(keeper *conditions.Keeper) {
flushing, cancel := context.WithTimeout(context.Background(), 10*time.Second)
defer cancel()
keeper.Close(flushing)
}
// watchTheMesh starts the watchdogs, the bus's advisories and the self-check over the condition store's keeper, for the
// serving controller; the returned function stops them and waits for them to end. Nil when there is no keeper.
func watchTheMesh(ctx context.Context, open *stores, server *link.Server, bus link.OverNATS, keeper *conditions.Keeper) func() {
if keeper == nil {
return nil
}
logf := func(format string, args ...any) { fmt.Printf(format+"\n", args...) }
stopHearing, err := server.HearAdvisories(logf)
if err != nil {
@@ -566,26 +581,26 @@ func watchTheMesh(ctx context.Context, open *stores, server *link.Server, bus li
d := &doctor{open: open, js: server.JetStream(), keeper: keeper, teller: bus, watchdogs: w, host: host}
w.doctor = d
doctorFrom = d
// Its own group as well as serve's: its stop waits for what it started.
watching, stop := context.WithCancel(ctx)
go w.keep(watching)
go d.keep(watching)
watching, running := withServingGroup(watching)
goServing(watching, func() { w.keep(watching) })
goServing(watching, func() { d.keep(watching) })
// And the healers (novox/hq to-be 45 §7): what is open that a registered healer answers, repaired
// under the lease and the brake, every act said.
healers := newHealing(open, keeper, bus, server.JetStream())
go healers.keep(watching)
goServing(watching, func() { healers.keep(watching) })
// And the asker (novox/hq ADR 0259): what needs the operator and names its answers is asked of them,
// and the answer chosen is performed on its warrant.
startAsking(watching, open, server, bus.Conn, keeper)
go forgettingOldHeals(watching, open.inventory)
goServing(watching, func() { forgettingOldHeals(watching, open.inventory) })
fmt.Printf("watching the mesh: %d signal(s) every %s, %d probe(s) every %s; what is wrong is kept in %s "+
"and said as %s events\n", len(watchedRows()), watchEvery, len(runnableProbes()), doctorEvery,
broker.ConditionsBucket, conditions.Seat)
return func() {
stop()
running.stop()
stopHearing()
flushing, cancel := context.WithTimeout(context.Background(), 10*time.Second)
defer cancel()
keeper.Close(flushing)
}
}