From e9a885cda925cd34f11671bcb5e7075648729658 Mon Sep 17 00:00:00 2001 From: jochen Date: Sun, 25 Oct 2026 16:23:55 +0100 Subject: [PATCH] Share the controller's stores with its take-ins and cap take-ins at 16 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. --- cmd/mesh-controller/compose_sdk.go | 13 ++- cmd/mesh-controller/composed_keys.go | 22 ++++- cmd/mesh-controller/composed_keys_cap_test.go | 58 +++++++++++++ cmd/mesh-controller/push.go | 7 +- cmd/mesh-controller/shared_stores_test.go | 86 +++++++++++++++++++ cmd/mesh-controller/stores.go | 38 +++++++- internal/store/ready_test.go | 82 ++++++++++++++++++ internal/store/store.go | 53 ++++++++++-- 8 files changed, 343 insertions(+), 16 deletions(-) create mode 100644 cmd/mesh-controller/composed_keys_cap_test.go create mode 100644 cmd/mesh-controller/shared_stores_test.go create mode 100644 internal/store/ready_test.go diff --git a/cmd/mesh-controller/compose_sdk.go b/cmd/mesh-controller/compose_sdk.go index 408ae511..ff8b2506 100644 --- a/cmd/mesh-controller/compose_sdk.go +++ b/cmd/mesh-controller/compose_sdk.go @@ -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) { diff --git a/cmd/mesh-controller/composed_keys.go b/cmd/mesh-controller/composed_keys.go index 8a934c69..8b8dd3ff 100644 --- a/cmd/mesh-controller/composed_keys.go +++ b/cmd/mesh-controller/composed_keys.go @@ -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). diff --git a/cmd/mesh-controller/composed_keys_cap_test.go b/cmd/mesh-controller/composed_keys_cap_test.go new file mode 100644 index 00000000..54303253 --- /dev/null +++ b/cmd/mesh-controller/composed_keys_cap_test.go @@ -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) + } +} diff --git a/cmd/mesh-controller/push.go b/cmd/mesh-controller/push.go index 651765ec..7c63572d 100644 --- a/cmd/mesh-controller/push.go +++ b/cmd/mesh-controller/push.go @@ -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 diff --git a/cmd/mesh-controller/shared_stores_test.go b/cmd/mesh-controller/shared_stores_test.go new file mode 100644 index 00000000..cfa93aa2 --- /dev/null +++ b/cmd/mesh-controller/shared_stores_test.go @@ -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) + } +} diff --git a/cmd/mesh-controller/stores.go b/cmd/mesh-controller/stores.go index 0fbd2703..cd3b1c33 100644 --- a/cmd/mesh-controller/stores.go +++ b/cmd/mesh-controller/stores.go @@ -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 +} diff --git a/internal/store/ready_test.go b/internal/store/ready_test.go new file mode 100644 index 00000000..5498adc0 --- /dev/null +++ b/internal/store/ready_test.go @@ -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) + } + } +} diff --git a/internal/store/store.go b/internal/store/store.go index d672bfbf..8ed3dbad 100644 --- a/internal/store/store.go +++ b/internal/store/store.go @@ -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) }