Share the controller's stores with its take-ins and cap take-ins at 16
mesh/delivery held for a person: merged without a passing check: only a person decides what follows
mesh/check checking e9a885cda925 on 62802cd96602
mesh/delivery held for a person: merged without a passing check: only a person decides what follows
mesh/check checking e9a885cda925 on 62802cd96602
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.
This commit is contained in:
@@ -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"
|
||||
"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) }
|
||||
|
||||
Reference in New Issue
Block a user