diff --git a/api/dbv1/db_replicas.go b/api/dbv1/db_replicas.go index bf4f1d28..f9ec66df 100644 --- a/api/dbv1/db_replicas.go +++ b/api/dbv1/db_replicas.go @@ -19,8 +19,8 @@ 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 { @@ -28,6 +28,9 @@ func NewDBPools(connectionStrings []string, logger *zap.Logger, env string, zapL if err != nil { return nil, err } + if maxConns > 0 { + connConfig.MaxConns = maxConns + } // Configure SQL logging if not in test environment if env != "test" { diff --git a/api/dbv1/db_replicas_test.go b/api/dbv1/db_replicas_test.go index 1e9cd275..bc368f1c 100644 --- a/api/dbv1/db_replicas_test.go +++ b/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) { diff --git a/api/server.go b/api/server.go index cb992caa..a1e22be2 100644 --- a/api/server.go +++ b/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)) } diff --git a/config/config.go b/config/config.go index 47e30739..b43c9f7e 100644 --- a/config/config.go +++ b/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 != "" { diff --git a/ddl/migrations/0233_comments_track_entity_created_at_idx.sql b/ddl/migrations/0233_comments_track_entity_created_at_idx.sql new file mode 100644 index 00000000..34a8f000 --- /dev/null +++ b/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; diff --git a/indexer/aggregates_calculator.go b/indexer/aggregates_calculator.go index c760b57b..0e8870b1 100644 --- a/indexer/aggregates_calculator.go +++ b/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) }