Reduce API read-replica recovery conflict storms

api/dbv1/db_replicas.go

@@ -19,15 +19,18 @@ type DBPools struct {
 
 // NewDBPools creates a new DBPools struct from a list of database connection strings.
 // It parses each connection string, configures SQL logging if not in test environment,
-// and creates connection pools for each replica.
-func NewDBPools(connectionStrings []string, logger *zap.Logger, env string, zapLevel zapcore.Level) (*DBPools, error) {
+// applies an optional per-replica connection limit, and creates connection pools.
+func NewDBPools(connectionStrings []string, maxConns int32, logger *zap.Logger, env string, zapLevel zapcore.Level) (*DBPools, error) {
 	var pools []*pgxpool.Pool
 
 	for _, connStr := range connectionStrings {
 		connConfig, err := pgxpool.ParseConfig(connStr)
 		if err != nil {
 			return nil, err
 		}
+		if maxConns > 0 {
+			connConfig.MaxConns = maxConns
+		}
 
 		// Configure SQL logging if not in test environment
 		if env != "test" {

api/dbv1/db_replicas_test.go

@@ -14,7 +14,7 @@ func TestNewDBPools(t *testing.T) {
 	logger := zap.NewNop()
 
 	// Test with empty connection strings
-	pools, err := NewDBPools([]string{}, logger, "test", zapcore.InfoLevel)
+	pools, err := NewDBPools([]string{}, 0, logger, "test", zapcore.InfoLevel)
 	if err != nil {
 		t.Fatalf("Expected no error for empty connection strings, got: %v", err)
 	}
@@ -23,10 +23,25 @@ func TestNewDBPools(t *testing.T) {
 	}
 
 	// Test with invalid connection string
-	_, err = NewDBPools([]string{"invalid://connection"}, logger, "test", zapcore.InfoLevel)
+	_, err = NewDBPools([]string{"invalid://connection"}, 0, logger, "test", zapcore.InfoLevel)
 	if err == nil {
 		t.Error("Expected error for invalid connection string, got nil")
 	}
+
+	pools, err = NewDBPools(
+		[]string{"postgresql://user:password@localhost/database"},
+		8,
+		logger,
+		"test",
+		zapcore.InfoLevel,
+	)
+	if err != nil {
+		t.Fatalf("Expected no error for valid connection string, got: %v", err)
+	}
+	defer pools.Close()
+	if got := pools.Replicas[0].Config().MaxConns; got != 8 {
+		t.Errorf("Expected max connections to be 8, got %d", got)
+	}
 }
 
 func TestChooseReplica(t *testing.T) {

api/server.go

@@ -72,7 +72,7 @@ func NewApiServer(config config.Config) *ApiServer {
 		connectionStrings = []string{config.ReadDbUrl}
 	}
 
-	pool, err := dbv1.NewDBPools(connectionStrings, logger, config.Env, config.ZapLevel)
+	pool, err := dbv1.NewDBPools(connectionStrings, config.ReadDbMaxConns, logger, config.Env, config.ZapLevel)
 	if err != nil {
 		logger.Fatal("read db connect failed", zap.Error(err))
 	}

config/config.go

@@ -20,6 +20,7 @@ type Config struct {
 	ZapLevel                       zapcore.Level
 	ReadDbUrl                      string
 	ReadDbReplicas                 []string
+	ReadDbMaxConns                 int32
 	WriteDbUrl                     string
 	RunMigrations                  bool
 	EsUrl                          string
@@ -54,15 +55,15 @@ type Config struct {
 	// Audius DelegateManager address — used to read
 	// getTotalDelegatorStake(holder).
 	EthDelegateManagerContractAddress string
-	SolanaIndexerWorkers         int
-	SolanaIndexerRetryInterval   time.Duration
-	CommsMessagePush             bool
-	AudiusdChainID               uint
-	AudiusdEntityManagerAddress  string
-	AudiusAppUrl                 string
-	RewardCodeAuthorizedKeys     []string
-	LaunchpadDeterministicSecret string
-	UnsplashKeys                 []string
+	SolanaIndexerWorkers              int
+	SolanaIndexerRetryInterval        time.Duration
+	CommsMessagePush                  bool
+	AudiusdChainID                    uint
+	AudiusdEntityManagerAddress       string
+	AudiusAppUrl                      string
+	RewardCodeAuthorizedKeys          []string
+	LaunchpadDeterministicSecret      string
+	UnsplashKeys                      []string
 	// Nodes that volunteer as STORE_ALL nodes and are always included in mirrors lists
 	StoreAllNodes []string
 	// Nodes that are truly dead and should not be included in rendezvous
@@ -102,6 +103,7 @@ var Cfg = Config{
 	LogLevel:                              os.Getenv("logLevel"),
 	ReadDbUrl:                             os.Getenv("readDbUrl"),
 	ReadDbReplicas:                        strings.Split(os.Getenv("readDbReplicas"), ","),
+	ReadDbMaxConns:                        8,
 	WriteDbUrl:                            os.Getenv("writeDbUrl"),
 	RunMigrations:                         os.Getenv("runMigrations") == "true",
 	EsUrl:                                 os.Getenv("elasticsearchUrl"),
@@ -311,6 +313,14 @@ func init() {
 		Cfg.CommsMessagePush = commsMessagePushEnabled
 	}
 
+	if v := os.Getenv("readDbMaxConns"); v != "" {
+		parsed, err := strconv.ParseInt(v, 10, 32)
+		if err != nil || parsed <= 0 {
+			log.Fatalf("Invalid readDbMaxConns %q: must be a positive integer", v)
+		}
+		Cfg.ReadDbMaxConns = int32(parsed)
+	}
+
 	// Solana indexer config
 	retryInterval := os.Getenv("solanaIndexerRetryInterval")
 	if retryInterval != "" {

ddl/migrations/0233_comments_track_entity_created_at_idx.sql

@@ -0,0 +1,15 @@
+-- Supports track comment listing and counts:
+--
+--   WHERE entity_id = ?
+--     AND entity_type = 'Track'
+--     AND is_delete = false
+--   ORDER BY created_at DESC
+--
+-- These endpoints otherwise scan the full comments table for each track. The
+-- included columns cover the comment fields used before moderation joins and
+-- keep this partial index small enough for the serving read path.
+CREATE INDEX CONCURRENTLY IF NOT EXISTS comments_track_entity_created_at_idx
+    ON public.comments USING btree (entity_id, created_at DESC)
+    INCLUDE (comment_id, user_id)
+    WHERE entity_type = 'Track'
+      AND is_delete = false;

indexer/aggregates_calculator.go

@@ -24,7 +24,7 @@ const aggregateScoreUpdateInterval = 10 * time.Minute
 
 func NewAggregatesCalculator(config config.Config) *AggregatesCalculator {
 	logger := logging.NewZapLogger(config).Named("AggregatesCalculator")
-	readPool, err := dbv1.NewDBPools([]string{config.ReadDbUrl}, logger, config.Env, config.ZapLevel)
+	readPool, err := dbv1.NewDBPools([]string{config.ReadDbUrl}, config.ReadDbMaxConns, logger, config.Env, config.ZapLevel)
 	if err != nil {
 		panic(err)
 	}