fix: protect publish path from mysql exhaustion

server/db/mysql/adapter.go

@@ -21,6 +21,7 @@ import (
 	"github.com/tinode/chat/server/auth"
 	dbadapter "github.com/tinode/chat/server/db"
 	"github.com/tinode/chat/server/db/common"
+	immetrics "github.com/tinode/chat/server/metrics"
 	"github.com/tinode/chat/server/store"
 	t "github.com/tinode/chat/server/store/types"
 )
@@ -109,6 +110,9 @@ type configType struct {
 	MaxOpenConns int `json:"max_open_conns,omitempty"`
 	// Maximum number of connections in the idle connection pool.
 	MaxIdleConns int `json:"max_idle_conns,omitempty"`
+	// Registry database has a separate, much smaller workload and pool budget.
+	RegistryMaxOpenConns int `json:"registry_max_open_conns,omitempty"`
+	RegistryMaxIdleConns int `json:"registry_max_idle_conns,omitempty"`
 	// Maximum amount of time a connection may be reused (in seconds).
 	ConnMaxLifetime int `json:"conn_max_lifetime,omitempty"`
 
@@ -185,6 +189,7 @@ func openMySQLDB(config configType, dsn, dbName string) (*sqlx.DB, string, error
 	if config.ConnMaxLifetime > 0 {
 		db.SetConnMaxLifetime(time.Duration(config.ConnMaxLifetime) * time.Second)
 	}
+	immetrics.RegisterDBPool(dbName, db.DB)
 	return db, dbName, nil
 }
 
@@ -281,7 +286,14 @@ func (a *adapter) Open(jsonconfig json.RawMessage) error {
 			a.txTimeout = time.Duration(float64(config.SqlTimeout)*txTimeoutMultiplier) * time.Second
 		}
 		if a.tenantRegistryDB != "" && a.tenantRegistryDB != a.dbName {
-			a.registryDB, _, err = openMySQLDB(config, a.dsn, a.tenantRegistryDB)
+			registryConfig := config
+			if config.RegistryMaxOpenConns > 0 {
+				registryConfig.MaxOpenConns = config.RegistryMaxOpenConns
+			}
+			if config.RegistryMaxIdleConns > 0 {
+				registryConfig.MaxIdleConns = config.RegistryMaxIdleConns
+			}
+			a.registryDB, _, err = openMySQLDB(registryConfig, a.dsn, a.tenantRegistryDB)
 			if err == nil {
 				err = a.registryDB.Ping()
 			}

server/im_security_policy_runtime.go

@@ -47,6 +47,12 @@ func imYouthModePolicyDecision(tenantID types.TenantID, actor types.Uid, action
 	if store.Security == nil || actor.IsZero() {
 		return imAllowDecision()
 	}
+	// Message content filtering is enforced by moderation below the policy
+	// check. Youth mode currently has no message-level deny rule, so avoid a
+	// database lookup on every ordinary publish.
+	if strings.TrimSpace(action) == imYouthActionMessage {
+		return imAllowDecision()
+	}
 	settings, err := store.Security.GetYouthMode(tenantID, actor)
 	if err != nil {
 		if !errors.Is(err, types.ErrNotFound) && !errors.Is(err, types.ErrUnsupported) {
@@ -65,10 +71,6 @@ func imYouthModePolicyDecision(tenantID types.TenantID, actor types.Uid, action
 				"policy_version": settings.PolicyVersion,
 			})
 		}
-	case imYouthActionMessage:
-		// content_filter is enforced by routing messages through content moderation
-		// before persistence; the decision remains allow unless moderation rejects it.
-		return imAllowDecision()
 	}
 	return imAllowDecision()
 }

server/im_security_policy_runtime_test.go

@@ -33,3 +33,18 @@ func TestYouthModeNightQuietRestrictsCalls(t *testing.T) {
 		t.Fatalf("daytime call should be allowed: %#v", decision)
 	}
 }
+
+func TestYouthModeOrdinaryMessageDoesNotReadDatabase(t *testing.T) {
+	oldSecurity := store.Security
+	security := &testSecurityStore{err: types.ErrInternal}
+	store.Security = security
+	defer func() { store.Security = oldSecurity }()
+
+	decision := imYouthModePolicyDecision(7, types.Uid(42), imYouthActionMessage, time.Now())
+	if !decision.Allowed {
+		t.Fatalf("ordinary message should continue to moderation: %#v", decision)
+	}
+	if security.youthModeReads != 0 {
+		t.Fatalf("ordinary message read youth settings %d times", security.youthModeReads)
+	}
+}

server/meetingim.conf

@@ -84,8 +84,10 @@
         "dsn": "mysql://root:oWxz*2=sb),ouW%p@tcp(mysql.118im.com:30306)/meeting_im?parseTime=true&collation=utf8mb4_0900_ai_ci",
         "tenant_registry_db": "hotgo",
         "tenant_db_routing": true,
-        "max_open_conns": 64,
-        "max_idle_conns": 64,
+        "max_open_conns": 24,
+        "max_idle_conns": 8,
+        "registry_max_open_conns": 4,
+        "registry_max_idle_conns": 2,
         "conn_max_lifetime": 60,
         "sql_timeout": 10
       }

server/metrics/db_pool.go

@@ -0,0 +1,80 @@
+package metrics
+
+import (
+	"database/sql"
+	"strings"
+	"sync"
+
+	"github.com/prometheus/client_golang/prometheus"
+)
+
+type dbStatsProvider interface {
+	Stats() sql.DBStats
+}
+
+type dbPoolCollector struct {
+	pools sync.Map
+	desc  map[string]*prometheus.Desc
+}
+
+func newDBPoolCollector() *dbPoolCollector {
+	labels := []string{"pod", "pool"}
+	return &dbPoolCollector{desc: map[string]*prometheus.Desc{
+		"max_open":            prometheus.NewDesc("im_db_pool_max_open_connections", "Configured maximum open database connections.", labels, nil),
+		"open":                prometheus.NewDesc("im_db_pool_open_connections", "Current open database connections.", labels, nil),
+		"in_use":              prometheus.NewDesc("im_db_pool_in_use_connections", "Current database connections in use.", labels, nil),
+		"idle":                prometheus.NewDesc("im_db_pool_idle_connections", "Current idle database connections.", labels, nil),
+		"wait":                prometheus.NewDesc("im_db_pool_wait_total", "Total waits for a database connection.", labels, nil),
+		"wait_duration":       prometheus.NewDesc("im_db_pool_wait_duration_seconds_total", "Total time spent waiting for database connections.", labels, nil),
+		"max_idle_closed":     prometheus.NewDesc("im_db_pool_max_idle_closed_total", "Connections closed because the idle limit was exceeded.", labels, nil),
+		"max_lifetime_closed": prometheus.NewDesc("im_db_pool_max_lifetime_closed_total", "Connections closed because their lifetime expired.", labels, nil),
+	}}
+}
+
+func (c *dbPoolCollector) Describe(ch chan<- *prometheus.Desc) {
+	for _, desc := range c.desc {
+		ch <- desc
+	}
+}
+
+func (c *dbPoolCollector) Collect(ch chan<- prometheus.Metric) {
+	pod := metricPodName()
+	c.pools.Range(func(key, value any) bool {
+		pool, ok := key.(string)
+		provider, providerOK := value.(dbStatsProvider)
+		if !ok || !providerOK {
+			return true
+		}
+		stats := provider.Stats()
+		labels := []string{pod, pool}
+		ch <- prometheus.MustNewConstMetric(c.desc["max_open"], prometheus.GaugeValue, float64(stats.MaxOpenConnections), labels...)
+		ch <- prometheus.MustNewConstMetric(c.desc["open"], prometheus.GaugeValue, float64(stats.OpenConnections), labels...)
+		ch <- prometheus.MustNewConstMetric(c.desc["in_use"], prometheus.GaugeValue, float64(stats.InUse), labels...)
+		ch <- prometheus.MustNewConstMetric(c.desc["idle"], prometheus.GaugeValue, float64(stats.Idle), labels...)
+		ch <- prometheus.MustNewConstMetric(c.desc["wait"], prometheus.CounterValue, float64(stats.WaitCount), labels...)
+		ch <- prometheus.MustNewConstMetric(c.desc["wait_duration"], prometheus.CounterValue, stats.WaitDuration.Seconds(), labels...)
+		ch <- prometheus.MustNewConstMetric(c.desc["max_idle_closed"], prometheus.CounterValue, float64(stats.MaxIdleClosed), labels...)
+		ch <- prometheus.MustNewConstMetric(c.desc["max_lifetime_closed"], prometheus.CounterValue, float64(stats.MaxLifetimeClosed), labels...)
+		return true
+	})
+}
+
+var dbPools = newDBPoolCollector()
+
+// RegisterDBPool exposes database/sql pool pressure for one logical database.
+// Registering the same name again replaces the provider, which keeps tests and
+// adapter reopen operations safe.
+func RegisterDBPool(name string, provider dbStatsProvider) {
+	if provider == nil {
+		return
+	}
+	name = strings.TrimSpace(name)
+	if name == "" {
+		name = "default"
+	}
+	dbPools.pools.Store(name, provider)
+}
+
+func init() {
+	Registry.MustRegister(dbPools)
+}

server/metrics/db_pool_test.go

@@ -0,0 +1,47 @@
+package metrics
+
+import (
+	"database/sql"
+	"testing"
+	"time"
+)
+
+type fixedDBStats struct {
+	stats sql.DBStats
+}
+
+func (s fixedDBStats) Stats() sql.DBStats { return s.stats }
+
+func TestDBPoolCollector(t *testing.T) {
+	RegisterDBPool("test", fixedDBStats{stats: sql.DBStats{
+		MaxOpenConnections: 32,
+		OpenConnections:    12,
+		InUse:              7,
+		Idle:               5,
+		WaitCount:          9,
+		WaitDuration:       1500 * time.Millisecond,
+	}})
+
+	families, err := Registry.Gather()
+	if err != nil {
+		t.Fatal(err)
+	}
+	want := map[string]bool{
+		"im_db_pool_max_open_connections":        false,
+		"im_db_pool_open_connections":            false,
+		"im_db_pool_in_use_connections":          false,
+		"im_db_pool_idle_connections":            false,
+		"im_db_pool_wait_total":                  false,
+		"im_db_pool_wait_duration_seconds_total": false,
+	}
+	for _, family := range families {
+		if _, ok := want[family.GetName()]; ok {
+			want[family.GetName()] = true
+		}
+	}
+	for name, found := range want {
+		if !found {
+			t.Errorf("metric %s was not gathered", name)
+		}
+	}
+}

server/store/group_message_runtime_test.go

@@ -17,6 +17,7 @@ type testGroupMessageRuntime struct {
 	shortStates     map[string]bool
 	shortValues     map[string]string
 	objects         map[string][]byte
+	sets            map[string]map[string]bool
 	deletedPrefixes []string
 	deletedObjects  []string
 }
@@ -28,6 +29,7 @@ func newTestGroupMessageRuntime() *testGroupMessageRuntime {
 		shortStates: make(map[string]bool),
 		shortValues: make(map[string]string),
 		objects:     make(map[string][]byte),
+		sets:        make(map[string]map[string]bool),
 	}
 }
 
@@ -186,16 +188,24 @@ func (r *testGroupMessageRuntime) DeleteCachedObjectsByPrefix(tenantID types.Ten
 	return nil
 }
 
-func (r *testGroupMessageRuntime) AddCachedSetMember(types.TenantID, string, string, string) error {
+func (r *testGroupMessageRuntime) AddCachedSetMember(tenantID types.TenantID, namespace, id, member string) error {
+	key := strconv.FormatInt(int64(tenantID), 10) + ":" + namespace + ":" + id
+	if r.sets[key] == nil {
+		r.sets[key] = make(map[string]bool)
+	}
+	r.sets[key][member] = true
 	return nil
 }
 
-func (r *testGroupMessageRuntime) RemoveCachedSetMember(types.TenantID, string, string, string) error {
+func (r *testGroupMessageRuntime) RemoveCachedSetMember(tenantID types.TenantID, namespace, id, member string) error {
+	key := strconv.FormatInt(int64(tenantID), 10) + ":" + namespace + ":" + id
+	delete(r.sets[key], member)
 	return nil
 }
 
-func (r *testGroupMessageRuntime) CachedSetContains(types.TenantID, string, string, string) (bool, error) {
-	return false, nil
+func (r *testGroupMessageRuntime) CachedSetContains(tenantID types.TenantID, namespace, id, member string) (bool, error) {
+	key := strconv.FormatInt(int64(tenantID), 10) + ":" + namespace + ":" + id
+	return r.sets[key][member], nil
 }
 
 func (r *testGroupMessageRuntime) AllowRate(types.TenantID, string, int, time.Duration) (bool, int, time.Duration, error) {
@@ -355,6 +365,29 @@ func TestMessageRuntimeSeqAndClientMidIdempotency(t *testing.T) {
 	}
 }
 
+func TestCanSendP2PUsesWarmRelationshipCache(t *testing.T) {
+	oldRuntime := groupMsgRuntime
+	runtime := newTestGroupMessageRuntime()
+	groupMsgRuntime = runtime
+	t.Cleanup(func() { groupMsgRuntime = oldRuntime })
+
+	tenantID := types.TenantID(7)
+	userID := types.Uid(42)
+	peerID := types.Uid(43)
+	cacheFriendPair(tenantID, userID, peerID)
+
+	allowed, relationship, err := (relationsMapper{}).CanSendP2P(tenantID, userID, peerID)
+	if err != nil || !allowed || relationship != types.RelationshipFriend {
+		t.Fatalf("warm friend cache was not used: allowed=%v relationship=%q err=%v", allowed, relationship, err)
+	}
+
+	cacheBlockOneWay(tenantID, peerID, userID)
+	allowed, relationship, err = (relationsMapper{}).CanSendP2P(tenantID, userID, peerID)
+	if err != nil || allowed || relationship != types.RelationshipBlockedByPeer {
+		t.Fatalf("reverse block cache was not enforced: allowed=%v relationship=%q err=%v", allowed, relationship, err)
+	}
+}
+
 func TestBumpMessageSeqFloorNeverMovesBackward(t *testing.T) {
 	oldRuntime := groupMsgRuntime
 	runtime := newTestGroupMessageRuntime()

server/store/store.go

@@ -3147,6 +3147,29 @@ func (relationsMapper) GetRelationshipContext(tenantID types.TenantID, userID, p
 }
 
 func (relationsMapper) CanSendP2P(tenantID types.TenantID, userID, peerID types.Uid) (bool, string, error) {
+	if userID.IsZero() || peerID.IsZero() || userID == peerID {
+		return false, "", types.ErrMalformed
+	}
+	// A warm relationship cache is sufficient for the publish hot path. Block
+	// entries are checked first in both directions; friendship and block writes
+	// keep these sets coherent. Redis errors and cache misses fall back to the
+	// authoritative database path below.
+	blockCacheReady := true
+	if blocked, err := cachedBlockContains(tenantID, userID, peerID); err != nil {
+		blockCacheReady = false
+	} else if blocked {
+		return false, types.RelationshipBlockedByMe, nil
+	}
+	if blocked, err := cachedBlockContains(tenantID, peerID, userID); err != nil {
+		blockCacheReady = false
+	} else if blocked {
+		return false, types.RelationshipBlockedByPeer, nil
+	}
+	if blockCacheReady {
+		if friend, err := cachedFriendContains(tenantID, userID, peerID); err == nil && friend {
+			return true, types.RelationshipFriend, nil
+		}
+	}
 	ctx, err := (relationsMapper{}).GetRelationshipContext(tenantID, userID, peerID, "")
 	if err != nil {
 		return false, "", err
@@ -3168,6 +3191,9 @@ func (relationsMapper) BlockUser(tenantID types.TenantID, blockerID, blockedID t
 	block, err := tenantAdp.TenantUserBlockUpsert(tenantID, blockerID, blockedID, now)
 	if err == nil {
 		cacheBlockOneWay(tenantID, blockerID, blockedID)
+		// A positive friend cache must never outlive a new block. Removing both
+		// directions also makes a later block-cache eviction fall back to MySQL.
+		uncacheFriendPair(tenantID, blockerID, blockedID)
 	}
 	return block, err
 }

server/tinode.conf

@@ -84,8 +84,10 @@
         "dsn": "mysql://root:oWxz*2=sb),ouW%p@tcp(mysql.118im.com:30306)/meeting_im?parseTime=true&collation=utf8mb4_0900_ai_ci",
         "tenant_registry_db": "hotgo",
         "tenant_db_routing": true,
-        "max_open_conns": 64,
-        "max_idle_conns": 64,
+        "max_open_conns": 24,
+        "max_idle_conns": 8,
+        "registry_max_open_conns": 4,
+        "registry_max_idle_conns": 2,
         "conn_max_lifetime": 60,
         "sql_timeout": 10
       }

server/user_verification_test.go

@@ -272,6 +272,7 @@ type testSecurityStore struct {
 	consumedTarget      string
 	independentPassword *types.IndependentPassword
 	youthMode           *types.YouthModeSettings
+	youthModeReads      int
 	attempts            int
 	lockedUntil         *time.Time
 	notifications       []types.SystemNotification
@@ -379,6 +380,7 @@ func (s *testSecurityStore) UpdateIndependentPasswordAttempts(_ types.TenantID,
 }
 
 func (s *testSecurityStore) GetYouthMode(_ types.TenantID, uid types.Uid) (*types.YouthModeSettings, error) {
+	s.youthModeReads++
 	if s.err != nil {
 		return nil, s.err
 	}