Compare commits
4
Commits
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
ef10a00f6d | ||
|
|
e9a885cda9 | ||
|
|
2814262ac0 | ||
|
|
0a536d7b6d |
@@ -9,6 +9,7 @@ import (
|
|||||||
"regexp"
|
"regexp"
|
||||||
"slices"
|
"slices"
|
||||||
"strings"
|
"strings"
|
||||||
|
"sync"
|
||||||
"time"
|
"time"
|
||||||
|
|
||||||
"git.novox.be/novox/mesh-sdk/go/catalogue"
|
"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{}}
|
f := &sourceForms{bases: map[string]string{}}
|
||||||
var failedAt time.Time // when the last read failed; zero when it did not
|
var failedAt time.Time // when the last read failed; zero when it did not
|
||||||
var failure error
|
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) {
|
f.read = func(ctx context.Context, seat string, again bool) (string, error) {
|
||||||
|
mu.Lock()
|
||||||
|
defer mu.Unlock()
|
||||||
if b, known := f.bases[seat]; known {
|
if b, known := f.bases[seat]; known {
|
||||||
return b, nil
|
return b, nil
|
||||||
}
|
}
|
||||||
|
|||||||
@@ -326,9 +326,13 @@ func restOfTheMesh(ctx context.Context, open *stores, shelf map[string]catalogue
|
|||||||
return world, nil
|
return world, nil
|
||||||
}
|
}
|
||||||
|
|
||||||
// restOfTheMeshOf is restOfTheMesh for a caller that holds the inventory alone: the other stores the Input reads are opened for
|
// restOfTheMeshOf is restOfTheMesh for a caller that holds the inventory alone: on the serving controller's stores where it
|
||||||
// the call and let go of again.
|
// 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) {
|
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}
|
open := &stores{inventory: inv}
|
||||||
defer func() {
|
defer func() {
|
||||||
if open.licences != nil {
|
if open.licences != nil {
|
||||||
@@ -338,9 +342,12 @@ func restOfTheMeshOf(ctx context.Context, inv *inventory.Inventory, shelf map[st
|
|||||||
open.identity.Close()
|
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
|
// 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.
|
// that two claimants could take, the bus's users left out for want of a credential.
|
||||||
func sayFound(ctx context.Context, d compose.Declared) {
|
func sayFound(ctx context.Context, d compose.Declared) {
|
||||||
|
|||||||
@@ -140,4 +140,35 @@ func TestTheDeclarationsStreamGoesOnceEveryNodeRunsTheNewEngine(t *testing.T) {
|
|||||||
if !declarationsRetiredNow(ctx, m.open.inventory) {
|
if !declarationsRetiredNow(ctx, m.open.inventory) {
|
||||||
t.Errorf("a raise after the retirement would assert the stream again")
|
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)
|
||||||
|
}
|
||||||
}
|
}
|
||||||
|
|||||||
@@ -163,7 +163,7 @@ func followDoneKey(ctx context.Context, open *stores, module string) {
|
|||||||
saidWait := ""
|
saidWait := ""
|
||||||
for {
|
for {
|
||||||
doneKeysLookAgain.Delete(module) // what this look reads covers every wake before it
|
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)
|
o := doneKeyObservation(module)
|
||||||
if err == nil {
|
if err == nil {
|
||||||
if keeper := conditionsFrom; keeper != nil && failed {
|
if keeper := conditionsFrom; keeper != nil && failed {
|
||||||
@@ -348,9 +348,25 @@ func doneKeysAtStart(ctx context.Context, modules func(context.Context) ([]strin
|
|||||||
wg.Wait()
|
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
|
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,
|
// 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.
|
// 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) {
|
func takeDoneKeyAtStart(ctx context.Context, open *stores, module string) {
|
||||||
@@ -361,7 +377,7 @@ func takeDoneKeyAtStart(ctx context.Context, open *stores, module string) {
|
|||||||
return
|
return
|
||||||
}
|
}
|
||||||
doneKeysLookAgain.Delete(module) // what this look reads covers every wake before it
|
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()
|
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
|
// 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).
|
// (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)
|
||||||
|
}
|
||||||
|
}
|
||||||
@@ -135,9 +135,18 @@ func declarationsDone(nodes []string, runs map[string]string, current string) bo
|
|||||||
return true
|
return true
|
||||||
}
|
}
|
||||||
|
|
||||||
// declarationsRetiredOn reads whether the declarations are done on the mesh now: every node's have of the node-engine
|
// declarationsRetiredOn reads whether the declarations are done on the mesh now: their stream gone from the bus, or
|
||||||
// against the fingerprint it is registered at.
|
// 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) {
|
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)
|
fps, err := currentFingerprints(ctx, inv)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
return false, err
|
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
|
// 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) {
|
func retireDeclarations(ctx context.Context, open *stores) (bool, error) {
|
||||||
if _, err := theLease.epoch(ctx); err != nil {
|
if _, err := theLease.epoch(ctx); err != nil {
|
||||||
return false, err
|
return false, err
|
||||||
@@ -324,8 +334,9 @@ func replacedComposition(in compose.Input, r replacedController) (sdkdelivery.Co
|
|||||||
return c, nil
|
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
|
// declarationsRetiredNow says the declarations are done on the mesh now, for a raise of the bus's objects and for what
|
||||||
// store it cannot read is not done, so the stream is asserted as before.
|
// 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 {
|
func declarationsRetiredNow(ctx context.Context, inv *inventory.Inventory) bool {
|
||||||
bus, err := aBus()
|
bus, err := aBus()
|
||||||
if err != nil {
|
if err != nil {
|
||||||
|
|||||||
@@ -3,6 +3,7 @@ package main
|
|||||||
import (
|
import (
|
||||||
"context"
|
"context"
|
||||||
"encoding/json"
|
"encoding/json"
|
||||||
|
"fmt"
|
||||||
"strings"
|
"strings"
|
||||||
"testing"
|
"testing"
|
||||||
|
|
||||||
@@ -110,3 +111,28 @@ func TestTheDeclarationsGoOnlyOnceEveryNodeRunsTheNewEngine(t *testing.T) {
|
|||||||
t.Errorf("done with no fingerprint registered, or no node")
|
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)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|||||||
@@ -13,6 +13,7 @@ import (
|
|||||||
"path"
|
"path"
|
||||||
"sort"
|
"sort"
|
||||||
"strings"
|
"strings"
|
||||||
|
"sync"
|
||||||
"time"
|
"time"
|
||||||
|
|
||||||
"git.novox.be/novox/mesh-sdk/go/buildstep"
|
"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.
|
// 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,
|
// 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.
|
// 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
|
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
|
||||||
|
|||||||
@@ -59,6 +59,9 @@ func serve(ctx context.Context) (err error) {
|
|||||||
}
|
}
|
||||||
defer open.Close()
|
defer open.Close()
|
||||||
inv := open.inventory
|
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
|
// **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
|
// 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.
|
// 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)
|
"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 {
|
if err != nil {
|
||||||
return err
|
return err
|
||||||
}
|
}
|
||||||
defer ident.Close()
|
|
||||||
|
|
||||||
// Established at start rather than on first use. A control plane that cannot sign is one
|
// 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
|
// whose declarations every node correctly refuses, and that should be a startup failure
|
||||||
|
|||||||
@@ -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)
|
||||||
|
}
|
||||||
|
}
|
||||||
@@ -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)
|
||||||
|
}
|
||||||
|
}
|
||||||
@@ -4,6 +4,7 @@ import (
|
|||||||
"context"
|
"context"
|
||||||
"fmt"
|
"fmt"
|
||||||
"os"
|
"os"
|
||||||
|
"sync"
|
||||||
"time"
|
"time"
|
||||||
|
|
||||||
"git.novox.be/novox/mesh-sdk/go/catalogue"
|
"git.novox.be/novox/mesh-sdk/go/catalogue"
|
||||||
@@ -163,6 +164,13 @@ func openLicences(ctx context.Context) (*licences.Licences, error) {
|
|||||||
return held, nil
|
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.
|
// stores is what one command has open.
|
||||||
//
|
//
|
||||||
// **Opened once, not once per machine.** Working out what a machine should be reaches the identity
|
// **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
|
inventory *inventory.Inventory
|
||||||
identity *identity.Identity
|
identity *identity.Identity
|
||||||
licences *licences.Licences
|
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.
|
// 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.
|
// 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) {
|
func (h *stores) Identity(ctx context.Context) (*identity.Identity, error) {
|
||||||
|
h.mu.Lock()
|
||||||
|
defer h.mu.Unlock()
|
||||||
if h.identity != nil {
|
if h.identity != nil {
|
||||||
return h.identity, nil
|
return h.identity, nil
|
||||||
}
|
}
|
||||||
opened, err := openIdentity(ctx)
|
opened, err := identityOpener(ctx)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
return nil, err
|
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.
|
// Licences is the context holding model access, opened if it has not been.
|
||||||
func (h *stores) Licences(ctx context.Context) (*licences.Licences, error) {
|
func (h *stores) Licences(ctx context.Context) (*licences.Licences, error) {
|
||||||
|
h.mu.Lock()
|
||||||
|
defer h.mu.Unlock()
|
||||||
if h.licences != nil {
|
if h.licences != nil {
|
||||||
return h.licences, nil
|
return h.licences, nil
|
||||||
}
|
}
|
||||||
opened, err := openLicences(ctx)
|
opened, err := licencesOpener(ctx)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
return nil, err
|
return nil, err
|
||||||
}
|
}
|
||||||
@@ -226,3 +241,22 @@ func (h *stores) Close() {
|
|||||||
h.inventory.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
|
||||||
|
}
|
||||||
|
|||||||
@@ -512,7 +512,9 @@ func (w *watchdogs) lostConsumers(ctx context.Context, heard []link.Advisory) (m
|
|||||||
for _, a := range deleted {
|
for _, a := range deleted {
|
||||||
_, err := w.js.Context().ConsumerInfo(a.Stream, a.Consumer)
|
_, err := w.js.Context().ConsumerInfo(a.Stream, a.Consumer)
|
||||||
switch {
|
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]
|
out[a.Stream+"."+a.Consumer] = want[a.Stream+"."+a.Consumer]
|
||||||
case err != nil:
|
case err != nil:
|
||||||
return nil, fmt.Errorf("whether %s exists cannot be read: %w", link.ConsumerInWords(a.Stream, a.Consumer), err)
|
return nil, fmt.Errorf("whether %s exists cannot be read: %w", link.ConsumerInWords(a.Stream, a.Consumer), err)
|
||||||
|
|||||||
@@ -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
|
// 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
|
// 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
|
// 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"
|
const DeclarationsStream = "NODES"
|
||||||
|
|
||||||
// RaiseAfterDeclarations is Raise once the declarations' stream is retired: the mesh's streams and consumers but the
|
// RaiseAfterDeclarations is Raise once the declarations' stream is retired: the mesh's streams and consumers but the
|
||||||
|
|||||||
@@ -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
@@ -24,6 +24,7 @@ import (
|
|||||||
"strings"
|
"strings"
|
||||||
"time"
|
"time"
|
||||||
|
|
||||||
|
"github.com/jackc/pgx/v5/pgconn"
|
||||||
"github.com/jackc/pgx/v5/pgxpool"
|
"github.com/jackc/pgx/v5/pgxpool"
|
||||||
|
|
||||||
"github.com/novox/mesh-controller/internal/envfile"
|
"github.com/novox/mesh-controller/internal/envfile"
|
||||||
@@ -213,7 +214,7 @@ func (s *Store) Pool() *pgxpool.Pool { return s.pool }
|
|||||||
|
|
||||||
// Close releases the connections.
|
// Close releases the connections.
|
||||||
func (s *Store) Close() {
|
func (s *Store) Close() {
|
||||||
if s.pool != nil {
|
if s != nil && s.pool != nil {
|
||||||
s.pool.Close()
|
s.pool.Close()
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
@@ -227,8 +228,10 @@ func (s *Store) Close() {
|
|||||||
func (s *Store) Ready(ctx context.Context, within time.Duration) error {
|
func (s *Store) Ready(ctx context.Context, within time.Duration) error {
|
||||||
deadline := time.Now().Add(within)
|
deadline := time.Now().Add(within)
|
||||||
var last error
|
var last error
|
||||||
|
wait := 250 * time.Millisecond
|
||||||
|
saidFull := false
|
||||||
for {
|
for {
|
||||||
err := s.pool.Ping(ctx)
|
err := pingStore(ctx, s)
|
||||||
if err == nil {
|
if err == nil {
|
||||||
return nil
|
return nil
|
||||||
}
|
}
|
||||||
@@ -240,10 +243,48 @@ func (s *Store) Ready(ctx context.Context, within time.Duration) error {
|
|||||||
return fmt.Errorf(
|
return fmt.Errorf(
|
||||||
"the %s store did not answer within %s: %w", s.context, within, last)
|
"the %s store did not answer within %s: %w", s.context, within, last)
|
||||||
}
|
}
|
||||||
select {
|
// A full store (SQLSTATE 53300) is not asked again every 250 ms: each ask is one more connection it refuses
|
||||||
case <-ctx.Done():
|
// while it is full. The wait doubles up to fullStoreMostWait, and the journal says once why (issues 1945 and
|
||||||
return errors.Join(ctx.Err(), last)
|
// 1277, 2026-10-25 16:08: the controller's own retries kept the store full).
|
||||||
case <-time.After(250 * time.Millisecond):
|
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) }
|
||||||
|
|||||||
Reference in New Issue
Block a user