Compare commits

..
Author SHA1 Message Date
jschoubben ef10a00f6d Share one reading of the forge's address across take-ins
Each take-in made a fresh set of source forms (newSourceForms), so each
read the whole mesh once to learn the forge's address: about 120 reads
at once at 16:08, each one more load on a store that was already full.
The take-ins of an inventory now share one reading, made again once it
is a minute old so a moved forge is read within that bound. A base that
could not be read is not kept. The forms' read takes a lock, since the
take-ins that share it run side by side.

Issues 1945 and 1277.
2026-10-25 16:24:52 +01:00
jschoubben e9a885cda9 Share the controller's stores with its take-ins and cap take-ins at 16
mesh/check checking e9a885cda925 on 62802cd96602
mesh/delivery held for a person: merged without a passing check: only a person decides what follows
At 16:07:36 the builder rewrote about 120 done keys at once. Each woke a
take-in of its own with no limit, and each take-in read the rest of the
mesh through restOfTheMeshOf, which opened a licences and an identity
pool per call. novox's store ran out of connections at 16:08 and refused
everyone, the controller's retries keeping it full.

- The serving controller lends its stores by inventory (shareStores);
  restOfTheMeshOf reads on them instead of opening pools per call, and a
  verb run as its own process still opens its own. stores opens each
  store once under a lock, since readers now share it. serve uses the
  shared identity store instead of a second pool.
- The watch's take-ins and the start's look share one cap of 16 take-ins
  at once (takeInSlots); a take-in waiting for its next try holds none.
- store.Ready backs off on a full store (SQLSTATE 53300), doubling its
  wait up to 5 s, and says once that the store is full.

Issues 1945 and 1277.
2026-10-25 16:23:55 +01:00
mesh-tools 2814262ac0 Merge pull request 'Keep the declarations retired once their stream is gone, so the bus is not called unhealthy for a stream release 6 deleted (issue 1945)' (#462) from fix/r6-nodes-stream-gone into main 2026-10-25 14:25:11 +00:00
jschoubben 0a536d7b6d Keep the declarations retired once their stream is gone, so nothing expects, asserts or makes NODES again (issue 1945)
mesh/check pass 0a536d7b6d79 on 2bcd85d2df4c | unit 966 run, 172 kept, 2 skipped | integration 1622 run, 236 kept, 7 skipped | over bound: integration…
mesh/delivery merged: source.main moved to novox/mesh-controller@2814262ac0fa
A node-engine newer than the registered one, or a node asleep on an older
have, made the done test false after the start deleted NODES; the bus's
health check, probes D6 and D7 and healer H3 then expected the stream and
every node's consumer again, raised them urgent and made them anew. Release
6 sends no declaration, so a stream made again helps no node: a gone stream
is retired. The advisory watchdog also reads a consumer whose stream is gone
as gone instead of failing.
2026-10-25 15:01:09 +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
23 changed files with 847 additions and 54 deletions
+4
View File
@@ -9,6 +9,7 @@ import (
"regexp"
"slices"
"strings"
"sync"
"time"
"git.novox.be/novox/mesh-sdk/go/catalogue"
@@ -107,7 +108,10 @@ func sourceFormsRead(ctx context.Context, now func() time.Time,
f := &sourceForms{bases: map[string]string{}}
var failedAt time.Time // when the last read failed; zero when it did not
var failure error
var mu sync.Mutex // the forms are shared by take-ins side by side (sharedSourceForms): one read at a time
f.read = func(ctx context.Context, seat string, again bool) (string, error) {
mu.Lock()
defer mu.Unlock()
if b, known := f.bases[seat]; known {
return b, nil
}
+10 -3
View File
@@ -326,9 +326,13 @@ func restOfTheMesh(ctx context.Context, open *stores, shelf map[string]catalogue
return world, nil
}
// restOfTheMeshOf is restOfTheMesh for a caller that holds the inventory alone: the other stores the Input reads are opened for
// the call and let go of again.
// restOfTheMeshOf is restOfTheMesh for a caller that holds the inventory alone: on the serving controller's stores where it
// lends them (sharedStores), never a pool per call; elsewhere (a verb run as its own process) the other stores the Input
// reads are opened for the call and let go of again.
func restOfTheMeshOf(ctx context.Context, inv *inventory.Inventory, shelf map[string]catalogue.Manifest, exclude string) (catalogue.World, error) {
if shared, ok := sharedStoresOf(inv); ok {
return restOfTheMeshOn(ctx, shared, shelf, exclude)
}
open := &stores{inventory: inv}
defer func() {
if open.licences != nil {
@@ -338,9 +342,12 @@ func restOfTheMeshOf(ctx context.Context, inv *inventory.Inventory, shelf map[st
open.identity.Close()
}
}()
return restOfTheMesh(ctx, open, shelf, exclude)
return restOfTheMeshOn(ctx, open, shelf, exclude)
}
// restOfTheMeshOn is restOfTheMesh; a variable so a test reads which stores it is handed.
var restOfTheMeshOn = restOfTheMesh
// sayFound says the lines a composition found while composing, once each: a machine left out of the rest of the mesh, a seat
// that two claimants could take, the bus's users left out for want of a credential.
func sayFound(ctx context.Context, d compose.Declared) {
@@ -140,4 +140,35 @@ func TestTheDeclarationsStreamGoesOnceEveryNodeRunsTheNewEngine(t *testing.T) {
if !declarationsRetiredNow(ctx, m.open.inventory) {
t.Errorf("a raise after the retirement would assert the stream again")
}
// Issue 1945: a node-engine other than the registered one landing on laptop (a newer build on its way) leaves the
// declarations retired, so what the health check, the probes and healer H3 expect of the bus names neither the
// stream nor a node's consumer on it, and asserting the bus's objects again does not make the stream anew.
have("laptop", fpOf('7'))
if !declarationsRetiredNow(ctx, m.open.inventory) {
t.Fatalf("with laptop on another node-engine after the stream went, the declarations are no longer retired")
}
streams, consumers, err := expectedBusObjects(ctx, m.open.inventory)
if err != nil {
t.Fatal(err)
}
for _, s := range streams {
if s.Name == broker.DeclarationsStream {
t.Errorf("the bus's health check and probe D7 still expect the stream %s", s.Name)
}
}
for _, c := range consumers {
if c.Stream == broker.DeclarationsStream {
t.Errorf("the bus's health check and probe D6 still expect %s on %s", c.Name, c.Stream)
}
}
bus, err := broker.Dial(m.conn.ConnectedUrl())
if err != nil {
t.Fatal(err)
}
t.Cleanup(bus.Close)
_ = assertOnSend(ctx, m.open.inventory, bus, " ")
if _, err := m.js.Stream(ctx, broker.DeclarationsStream); !errors.Is(err, jetstream.ErrStreamNotFound) {
t.Errorf("healer H3's assertion of the bus's objects made the stream anew: %v", err)
}
}
+19 -3
View File
@@ -163,7 +163,7 @@ func followDoneKey(ctx context.Context, open *stores, module string) {
saidWait := ""
for {
doneKeysLookAgain.Delete(module) // what this look reads covers every wake before it
err := doneKeyTakeOnce(ctx, open, module)
err := takeOnceInSlot(ctx, open, module)
o := doneKeyObservation(module)
if err == nil {
if keeper := conditionsFrom; keeper != nil && failed {
@@ -348,9 +348,25 @@ func doneKeysAtStart(ctx context.Context, modules func(context.Context) ([]strin
wg.Wait()
}
// startTakeIns is how many take-ins the start's look runs at once.
// 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()
}
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) {
@@ -361,7 +377,7 @@ func takeDoneKeyAtStart(ctx context.Context, open *stores, module string) {
return
}
doneKeysLookAgain.Delete(module) // what this look reads covers every wake before it
err := doneKeyTakeOnce(ctx, open, module)
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).
@@ -0,0 +1,58 @@
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)
}
}
+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)
}
+16 -5
View File
@@ -135,9 +135,18 @@ func declarationsDone(nodes []string, runs map[string]string, current string) bo
return true
}
// declarationsRetiredOn reads whether the declarations are done on the mesh now: every node's have of the node-engine
// against the fingerprint it is registered at.
// declarationsRetiredOn reads whether the declarations are done on the mesh now: their stream gone from the bus, or
// every node's have of the node-engine against the fingerprint it is registered at.
//
// **A stream that is gone stays gone** (issue 1945). Release 6 sends no declaration (ADR 0539 clauses 5 and 11), so a
// stream made again would hold no node's last declaration and help no node; and the done test alone is not lasting:
// a node-engine newer than the one registered landing, or a node asleep on an older have, makes it false again after
// the start deleted the stream, and then the raise, the bus's health check (H-bus), the probes (D6, D7) and healer H3
// expected the stream and every node's consumer again, raised them missing and made them anew (2026-10-25 14:52).
func declarationsRetiredOn(ctx context.Context, inv *inventory.Inventory, js jetstream.JetStream) (bool, error) {
if _, err := js.Stream(ctx, broker.DeclarationsStream); errors.Is(err, jetstream.ErrStreamNotFound) {
return true, nil
}
fps, err := currentFingerprints(ctx, inv)
if err != nil {
return false, err
@@ -162,7 +171,8 @@ func declarationsRetiredOn(ctx context.Context, inv *inventory.Inventory, js jet
}
// retireDeclarations deletes the declarations' stream, with every node's consumer on it, once every node runs the new
// node-engine, and says it once; a stream already gone is said by nobody. It answers whether the declarations are done.
// node-engine, and says it once; a stream already gone is said by nobody, and is done. It answers whether the
// declarations are done.
func retireDeclarations(ctx context.Context, open *stores) (bool, error) {
if _, err := theLease.epoch(ctx); err != nil {
return false, err
@@ -324,8 +334,9 @@ func replacedComposition(in compose.Input, r replacedController) (sdkdelivery.Co
return c, nil
}
// declarationsRetiredNow says the declarations are done on the mesh now, for a raise of the bus's objects; a bus or a
// store it cannot read is not done, so the stream is asserted as before.
// declarationsRetiredNow says the declarations are done on the mesh now, for a raise of the bus's objects and for what
// the health check, the probes and the healers expect of the bus; a bus or a store it cannot read is not done, so a
// stream still there is asserted as before.
func declarationsRetiredNow(ctx context.Context, inv *inventory.Inventory) bool {
bus, err := aBus()
if err != nil {
@@ -3,6 +3,7 @@ package main
import (
"context"
"encoding/json"
"fmt"
"strings"
"testing"
@@ -110,3 +111,28 @@ func TestTheDeclarationsGoOnlyOnceEveryNodeRunsTheNewEngine(t *testing.T) {
t.Errorf("done with no fingerprint registered, or no node")
}
}
// streamsGone is a bus that holds none of the streams asked for; every other part of it is unread.
type streamsGone struct{ jetstream.JetStream }
func (streamsGone) Stream(context.Context, string) (jetstream.Stream, error) {
return nil, jetstream.ErrStreamNotFound
}
// Issue 1945 (the alarm of a stream the start deleted): once the declarations' stream is gone from the bus, the
// declarations are retired whatever the nodes run now, so no raise, health check, probe or healer expects the stream
// or a node's consumer on it again. A node-engine newer than the registered one, or a node asleep on an older have,
// made the done test false after the stream went, and the bus's objects named it missing again (2026-10-25 14:52).
func TestTheDeclarationsStayRetiredOnceTheirStreamIsGone(t *testing.T) {
retired, err := func() (retired bool, err error) {
defer func() {
if p := recover(); p != nil {
err = fmt.Errorf("it read what the nodes run although the stream is gone: %v", p)
}
}()
return declarationsRetiredOn(t.Context(), nil, streamsGone{})
}()
if err != nil || !retired {
t.Fatalf("with no declarations' stream on the bus: retired %v (%v), want retired", retired, err)
}
}
+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")
}
}
@@ -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)
}
}
+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)
+39 -1
View File
@@ -13,6 +13,7 @@ import (
"path"
"sort"
"strings"
"sync"
"time"
"git.novox.be/novox/mesh-sdk/go/buildstep"
@@ -122,7 +123,7 @@ var predefinedAtTrunk = func(ctx context.Context, inv *inventory.Inventory) (pre
}
// formsOfSources reads how the sources are spelled on this mesh; a variable so a test names the forge's address.
var formsOfSources = newSourceForms
var formsOfSources = sharedSourceForms
// errNoCatalogueRepository says no module of the mesh is built from the catalogue's repository on the forge: not a wait,
// since no read of the forge makes one appear (issue 1902); the plan fails with these words.
@@ -632,3 +633,40 @@ func handCoversEvery(m catalogue.Manifest, k builderkeys.Key, hand map[string]st
}
return nil
}
// sharedSourceForms is the one reading of how the sources are spelled (the forge's address above all) that every take-in
// of the inventory shares, made again once it is older than sourceFormsKept, so a forge that moved is read within that
// bound. A fresh reading per take-in read the whole mesh once each, about 120 at once at 16:08 on 2026-10-25, each with
// pools of its own (issues 1945 and 1277). A base that could not be read is not kept: the reading reads it again.
func sharedSourceForms(ctx context.Context, inv *inventory.Inventory) *sourceForms {
sharedFormsMu.Lock()
defer sharedFormsMu.Unlock()
now := sourceFormsNow()
if kept, ok := sharedForms.Load(inv); ok {
if k := kept.(keptForms); now.Sub(k.at) < sourceFormsKept {
return k.forms
}
}
// Read with no caller's deadline: the reading outlives the take-in that made it.
forms := sourceFormsFor(context.WithoutCancel(ctx), inv)
sharedForms.Store(inv, keptForms{forms: forms, at: now})
return forms
}
// keptForms is a shared reading of the sources and when it was made.
type keptForms struct {
forms *sourceForms
at time.Time
}
// sourceFormsFor makes a fresh reading of the sources; a variable so a test counts them.
var sourceFormsFor = newSourceForms
var (
sharedForms sync.Map // *inventory.Inventory -> keptForms
sharedFormsMu sync.Mutex
sourceFormsNow = time.Now
)
// sourceFormsKept is how long the take-ins share one reading of the sources before it is made again.
const sourceFormsKept = time.Minute
+5 -2
View File
@@ -59,6 +59,9 @@ func serve(ctx context.Context) (err error) {
}
defer open.Close()
inv := open.inventory
// Lent to every reader in this process that holds only the inventory, so none opens pools of its own per call
// (issues 1945 and 1277: about 120 take-ins at once filled novox's store at 16:08 on 2026-10-25).
defer shareStores(open)()
// **What this build reads of the store's schema, on record** (novox/hq issue 352): the highest migration
// it carries, by its version, so a gate that would put this build back later knows it reads the store
// as it is then. A build that does not know its version records nothing, and is never put back.
@@ -75,11 +78,11 @@ func serve(ctx context.Context) (err error) {
"recorded, and a gate will never put it back\n", RunningBuildVar)
}
ident, err := openIdentity(ctx)
// The controller's own identity store, the one its readers share (closed with open), not a second pool.
ident, err := open.Identity(ctx)
if err != nil {
return err
}
defer ident.Close()
// Established at start rather than on first use. A control plane that cannot sign is one
// whose declarations every node correctly refuses, and that should be a startup failure
+52
View File
@@ -0,0 +1,52 @@
package main
import (
"context"
"sync"
"sync/atomic"
"testing"
"time"
"git.novox.be/novox/mesh-sdk/go/catalogue"
"github.com/novox/mesh-controller/internal/inventory"
)
// The take-ins share one reading of how the sources are spelled (the forge's address): a burst of 120 take-ins reads the
// mesh once, not once each (issues 1945 and 1277: each take-in's fresh read opened pools of its own at 16:08 on
// 2026-10-25). A reading older than sourceFormsKept is made again, so a moved forge is read.
func TestTakeInsShareOneReadingOfTheSources(t *testing.T) {
was, wasNow := sourceFormsFor, sourceFormsNow
t.Cleanup(func() { sourceFormsFor, sourceFormsNow = was, wasNow; sharedForms = sync.Map{} })
var made, worldReads atomic.Int64
sourceFormsFor = func(ctx context.Context, _ *inventory.Inventory) *sourceForms {
made.Add(1)
return sourceFormsRead(ctx, time.Now, func(context.Context) (catalogue.World, error) {
worldReads.Add(1)
return catalogue.World{}, nil
})
}
clock := time.Date(2026, 10, 25, 16, 8, 0, 0, time.UTC)
var clockMu sync.Mutex
sourceFormsNow = func() time.Time { clockMu.Lock(); defer clockMu.Unlock(); return clock }
inv := &inventory.Inventory{}
var wg sync.WaitGroup
for range 120 {
wg.Add(1)
go func() {
defer wg.Done()
f := formsOfSources(context.Background(), inv)
_, _ = f.read(context.Background(), gitSeat, true)
}()
}
wg.Wait()
if n := made.Load(); n != 1 {
t.Fatalf("made %d readings of the sources for 120 take-ins, want one", n)
}
clockMu.Lock()
clock = clock.Add(sourceFormsKept + time.Second)
clockMu.Unlock()
formsOfSources(context.Background(), inv)
if n := made.Load(); n != 2 {
t.Fatalf("a reading older than %s was kept (%d made)", sourceFormsKept, n)
}
}
+86
View File
@@ -0,0 +1,86 @@
package main
import (
"context"
"sync"
"sync/atomic"
"testing"
"git.novox.be/novox/mesh-sdk/go/catalogue"
"github.com/novox/mesh-controller/internal/identity"
"github.com/novox/mesh-controller/internal/inventory"
"github.com/novox/mesh-controller/internal/licences"
)
// The serving controller's reads of the rest of the mesh open no pool per call: the licences and identity stores are the
// controller's own, opened once and shared. On 2026-10-25 at 16:08 each take-in opened two pools of its own
// (restOfTheMeshOf), about 120 at once, and novox's store refused every new connection (issues 1945 and 1277).
func TestRestOfTheMeshOfOpensNoPoolPerCall(t *testing.T) {
wasIdent, wasLic, wasRest := identityOpener, licencesOpener, restOfTheMeshOn
t.Cleanup(func() { identityOpener, licencesOpener, restOfTheMeshOn = wasIdent, wasLic, wasRest })
var identities, licencesOpened atomic.Int64
identityOpener = func(context.Context) (*identity.Identity, error) {
identities.Add(1)
return &identity.Identity{}, nil
}
licencesOpener = func(context.Context) (*licences.Licences, error) {
licencesOpened.Add(1)
return &licences.Licences{}, nil
}
// The read of the Input reaches both stores, as readComposeInput does.
restOfTheMeshOn = func(ctx context.Context, open *stores, _ map[string]catalogue.Manifest, _ string) (catalogue.World, error) {
if _, err := open.Licences(ctx); err != nil {
return catalogue.World{}, err
}
if _, err := open.Identity(ctx); err != nil {
return catalogue.World{}, err
}
return catalogue.World{}, nil
}
inv := &inventory.Inventory{}
serving := &stores{inventory: inv}
unshare := shareStores(serving)
t.Cleanup(unshare)
var wg sync.WaitGroup
for range 120 {
wg.Add(1)
go func() {
defer wg.Done()
if _, err := restOfTheMeshOf(context.Background(), inv, nil, ""); err != nil {
t.Error(err)
}
}()
}
wg.Wait()
if n := licencesOpened.Load(); n != 1 {
t.Fatalf("opened the licences store %d times for 120 reads, want once", n)
}
if n := identities.Load(); n != 1 {
t.Fatalf("opened the identity store %d times for 120 reads, want once", n)
}
if serving.licences == nil || serving.identity == nil {
t.Fatal("the reads did not use the serving controller's stores")
}
}
// A process that shares no stores (a verb run as its own short process) still opens them for the call and lets go.
func TestRestOfTheMeshOfOpensItsOwnWhereNoneAreShared(t *testing.T) {
wasIdent, wasLic, wasRest := identityOpener, licencesOpener, restOfTheMeshOn
t.Cleanup(func() { identityOpener, licencesOpener, restOfTheMeshOn = wasIdent, wasLic, wasRest })
var opened atomic.Int64
licencesOpener = func(context.Context) (*licences.Licences, error) { opened.Add(1); return &licences.Licences{}, nil }
restOfTheMeshOn = func(ctx context.Context, open *stores, _ map[string]catalogue.Manifest, _ string) (catalogue.World, error) {
_, err := open.Licences(ctx)
return catalogue.World{}, err
}
inv := &inventory.Inventory{}
for range 2 {
if _, err := restOfTheMeshOf(context.Background(), inv, nil, ""); err != nil {
t.Fatal(err)
}
}
if n := opened.Load(); n != 2 {
t.Fatalf("opened %d times, want once per call", n)
}
}
+36 -2
View File
@@ -4,6 +4,7 @@ import (
"context"
"fmt"
"os"
"sync"
"time"
"git.novox.be/novox/mesh-sdk/go/catalogue"
@@ -163,6 +164,13 @@ func openLicences(ctx context.Context) (*licences.Licences, error) {
return held, nil
}
// identityOpener and licencesOpener open a store the first time a set of stores needs it; variables so a test counts
// the pools opened.
var (
identityOpener = openIdentity
licencesOpener = openLicences
)
// stores is what one command has open.
//
// **Opened once, not once per machine.** Working out what a machine should be reaches the identity
@@ -176,6 +184,9 @@ type stores struct {
inventory *inventory.Inventory
identity *identity.Identity
licences *licences.Licences
// mu keeps one opening of each store at a time: a serving controller's stores are shared by readers that run side
// by side (sharedStores).
mu sync.Mutex
}
// open connects to the inventory, which every command that touches the mesh needs.
@@ -189,10 +200,12 @@ func openStores(ctx context.Context) (*stores, error) {
// Identity is this control plane's own identity context, opened if it has not been.
func (h *stores) Identity(ctx context.Context) (*identity.Identity, error) {
h.mu.Lock()
defer h.mu.Unlock()
if h.identity != nil {
return h.identity, nil
}
opened, err := openIdentity(ctx)
opened, err := identityOpener(ctx)
if err != nil {
return nil, err
}
@@ -202,10 +215,12 @@ func (h *stores) Identity(ctx context.Context) (*identity.Identity, error) {
// Licences is the context holding model access, opened if it has not been.
func (h *stores) Licences(ctx context.Context) (*licences.Licences, error) {
h.mu.Lock()
defer h.mu.Unlock()
if h.licences != nil {
return h.licences, nil
}
opened, err := openLicences(ctx)
opened, err := licencesOpener(ctx)
if err != nil {
return nil, err
}
@@ -226,3 +241,22 @@ func (h *stores) Close() {
h.inventory.Close()
}
}
// sharedStores are the stores a serving controller lends, by its inventory, to every reader in the process that holds
// only the inventory (restOfTheMeshOf): one pool of each store for the controller, never one per call (issues 1945 and
// 1277, 2026-10-25 16:08: about 120 take-ins opened two pools each and filled novox's store).
var sharedStores sync.Map // *inventory.Inventory -> *stores
// shareStores lends open by its inventory until the function returned is called.
func shareStores(open *stores) func() {
sharedStores.Store(open.inventory, open)
return func() { sharedStores.CompareAndDelete(open.inventory, open) }
}
// sharedStoresOf are the stores lent for inv, if any are.
func sharedStoresOf(inv *inventory.Inventory) (*stores, bool) {
if open, ok := sharedStores.Load(inv); ok {
return open.(*stores), true
}
return nil, false
}
+3 -1
View File
@@ -512,7 +512,9 @@ func (w *watchdogs) lostConsumers(ctx context.Context, heard []link.Advisory) (m
for _, a := range deleted {
_, err := w.js.Context().ConsumerInfo(a.Stream, a.Consumer)
switch {
case errors.Is(err, nats.ErrConsumerNotFound):
// A consumer whose stream is gone is gone with it: the retired declarations' stream took every node's
// consumer, and its deletion said each as deleted (issue 1945).
case errors.Is(err, nats.ErrConsumerNotFound) || errors.Is(err, nats.ErrStreamNotFound):
out[a.Stream+"."+a.Consumer] = want[a.Stream+"."+a.Consumer]
case err != nil:
return nil, fmt.Errorf("whether %s exists cannot be read: %w", link.ConsumerInWords(a.Stream, a.Consumer), err)
+2 -1
View File
@@ -165,7 +165,8 @@ func HandPathWrites(nodes []string) ([]string, error) {
// DeclarationsStream is the stream of each node's last declaration, `mesh.node.<node>.declare` (issue 107). Release 6
// sends none (ADR 0539 clauses 5 and 11): the stream is kept, with each node's consumer, only so a node away during the
// switch finds its last declaration and with it the new node-engine; the controller deletes it once every node's have
// names the new node-engine, and from then on neither asserts it nor any node's consumer on it.
// names the new node-engine, and from then on neither asserts it nor any node's consumer on it; nor makes it again once
// it is gone from the bus for any reason, since a stream made anew would hold no node's last declaration (issue 1945).
const DeclarationsStream = "NODES"
// RaiseAfterDeclarations is Raise once the declarations' stream is retired: the mesh's streams and consumers but the
+82
View File
@@ -0,0 +1,82 @@
package store
import (
"context"
"strings"
"testing"
"time"
"github.com/jackc/pgx/v5/pgconn"
)
// A full store (SQLSTATE 53300, "too many clients already") is not asked again every 250 ms: Ready backs off,
// doubling its wait up to a bound, and says once that the store is full (issues 1945 and 1277, 2026-10-25 16:08: the
// controller's retries kept novox's store full).
func TestReadyBacksOffOnAFullStore(t *testing.T) {
wasPing, wasWait, wasSay := pingStore, readyWait, readySay
t.Cleanup(func() { pingStore, readyWait, readySay = wasPing, wasWait, wasSay })
asked := 0
pingStore = func(context.Context, *Store) error {
asked++
if asked <= 8 {
return &pgconn.PgError{Code: "53300", Message: "sorry, too many clients already"}
}
return nil
}
var waits []time.Duration
readyWait = func(_ context.Context, d time.Duration) error { waits = append(waits, d); return nil }
var said []string
readySay = func(line string) { said = append(said, line) }
if err := (&Store{context: "example"}).Ready(context.Background(), time.Hour); err != nil {
t.Fatalf("Ready: %v", err)
}
if len(waits) != 8 {
t.Fatalf("waited %d times, want 8: %v", len(waits), waits)
}
for i := 1; i < len(waits); i++ {
if waits[i] < waits[i-1] {
t.Fatalf("wait %d (%s) is shorter than the one before (%s): %v", i, waits[i], waits[i-1], waits)
}
}
if waits[1] <= waits[0] {
t.Fatalf("Ready does not back off on a full store: %v", waits)
}
if most := waits[len(waits)-1]; most > fullStoreMostWait {
t.Fatalf("the wait grew past its bound %s: %v", fullStoreMostWait, waits)
}
full := 0
for _, l := range said {
if strings.Contains(l, "full") {
full++
}
}
if full != 1 {
t.Fatalf("said that the store is full %d times, want once: %q", full, said)
}
}
// A store that does not answer for another reason (not up yet) is still asked again every 250 ms.
func TestReadyWaitsShortlyOnAStoreNotUp(t *testing.T) {
wasPing, wasWait, wasSay := pingStore, readyWait, readySay
t.Cleanup(func() { pingStore, readyWait, readySay = wasPing, wasWait, wasSay })
asked := 0
pingStore = func(context.Context, *Store) error {
asked++
if asked <= 3 {
return context.DeadlineExceeded
}
return nil
}
var waits []time.Duration
readyWait = func(_ context.Context, d time.Duration) error { waits = append(waits, d); return nil }
readySay = func(string) {}
if err := (&Store{context: "example"}).Ready(context.Background(), time.Hour); err != nil {
t.Fatalf("Ready: %v", err)
}
for _, w := range waits {
if w != 250*time.Millisecond {
t.Fatalf("waits %v, want 250ms each", waits)
}
}
}
+47 -6
View File
@@ -24,6 +24,7 @@ import (
"strings"
"time"
"github.com/jackc/pgx/v5/pgconn"
"github.com/jackc/pgx/v5/pgxpool"
"github.com/novox/mesh-controller/internal/envfile"
@@ -213,7 +214,7 @@ func (s *Store) Pool() *pgxpool.Pool { return s.pool }
// Close releases the connections.
func (s *Store) Close() {
if s.pool != nil {
if s != nil && s.pool != nil {
s.pool.Close()
}
}
@@ -227,8 +228,10 @@ func (s *Store) Close() {
func (s *Store) Ready(ctx context.Context, within time.Duration) error {
deadline := time.Now().Add(within)
var last error
wait := 250 * time.Millisecond
saidFull := false
for {
err := s.pool.Ping(ctx)
err := pingStore(ctx, s)
if err == nil {
return nil
}
@@ -240,10 +243,48 @@ func (s *Store) Ready(ctx context.Context, within time.Duration) error {
return fmt.Errorf(
"the %s store did not answer within %s: %w", s.context, within, last)
}
select {
case <-ctx.Done():
return errors.Join(ctx.Err(), last)
case <-time.After(250 * time.Millisecond):
// A full store (SQLSTATE 53300) is not asked again every 250 ms: each ask is one more connection it refuses
// while it is full. The wait doubles up to fullStoreMostWait, and the journal says once why (issues 1945 and
// 1277, 2026-10-25 16:08: the controller's own retries kept the store full).
if isFull(err) {
if !saidFull {
readySay(fmt.Sprintf("the %s store is full (too many clients already); asked again at a growing "+
"interval, at most every %s, for at most %s in all", s.context, fullStoreMostWait, within))
saidFull = true
}
} else {
wait = 250 * time.Millisecond
}
if err := readyWait(ctx, wait); err != nil {
return errors.Join(err, last)
}
if isFull(err) {
wait = min(wait*2, fullStoreMostWait)
}
}
}
// pingStore asks the store once whether it answers; a variable so a test needs no database.
var pingStore = func(ctx context.Context, s *Store) error { return s.pool.Ping(ctx) }
// readyWait waits d or until ctx ends; a variable so a test waits in no real time.
var readyWait = func(ctx context.Context, d time.Duration) error {
select {
case <-ctx.Done():
return ctx.Err()
case <-time.After(d):
return nil
}
}
// readySay says a line of Ready's wait in the journal; a variable so a test reads it.
// isFull says whether err is the store refusing a connection because it is full (SQLSTATE 53300).
func isFull(err error) bool {
var pg *pgconn.PgError
return errors.As(err, &pg) && pg.Code == "53300"
}
// fullStoreMostWait bounds the wait between two asks of a full store.
const fullStoreMostWait = 5 * time.Second
var readySay = func(line string) { fmt.Fprintln(os.Stderr, line) }