diff --git a/.gitea/workflows/deploy.yml b/.gitea/workflows/deploy.yml index 1ecb563..5270cb7 100644 --- a/.gitea/workflows/deploy.yml +++ b/.gitea/workflows/deploy.yml @@ -41,7 +41,11 @@ jobs: CLUSTER_PROXY_TO_MASTER_WORKERS=16 \ CLUSTER_PROXY_TO_MASTER_ENQUEUE_TIMEOUT_MS=300 \ CLUSTER_ROUTE_RETRIES=3 \ - CLUSTER_ROUTE_RETRY_BACKOFF_MS=25 + CLUSTER_ROUTE_RETRY_BACKOFF_MS=25 \ + MYSQL_MAX_OPEN_CONNS=24 \ + MYSQL_MAX_IDLE_CONNS=8 \ + MYSQL_REGISTRY_MAX_OPEN_CONNS=4 \ + MYSQL_REGISTRY_MAX_IDLE_CONNS=2 kubectl patch statefulset/im-server -n im-server-system --type=strategic -p \ '{"spec":{"template":{"spec":{"containers":[{"name":"im-server","readinessProbe":{"httpGet":{"path":"/healthz","port":6060},"timeoutSeconds":3,"periodSeconds":10,"failureThreshold":3},"livenessProbe":{"httpGet":{"path":"/livez","port":6060},"timeoutSeconds":3,"periodSeconds":20,"failureThreshold":3}}]}}}}' if kubectl auth can-i get services -n im-server-system >/dev/null 2>&1 && \ @@ -74,10 +78,20 @@ jobs: jsonpath='{.spec.template.spec.containers[?(@.name=="im-server")].resources.limits.memory}')" liveness="$(kubectl get statefulset/im-server -n im-server-system -o \ jsonpath='{.spec.template.spec.containers[?(@.name=="im-server")].livenessProbe.httpGet.path}')" - echo "effective requests=${request_cpu}/${request_memory} limits=${limit_cpu}/${limit_memory} liveness=${liveness}" + mysql_max_open="$(kubectl get statefulset/im-server -n im-server-system -o \ + jsonpath='{.spec.template.spec.containers[?(@.name=="im-server")].env[?(@.name=="MYSQL_MAX_OPEN_CONNS")].value}')" + mysql_max_idle="$(kubectl get statefulset/im-server -n im-server-system -o \ + jsonpath='{.spec.template.spec.containers[?(@.name=="im-server")].env[?(@.name=="MYSQL_MAX_IDLE_CONNS")].value}')" + mysql_registry_max_open="$(kubectl get statefulset/im-server -n im-server-system -o \ + jsonpath='{.spec.template.spec.containers[?(@.name=="im-server")].env[?(@.name=="MYSQL_REGISTRY_MAX_OPEN_CONNS")].value}')" + mysql_registry_max_idle="$(kubectl get statefulset/im-server -n im-server-system -o \ + jsonpath='{.spec.template.spec.containers[?(@.name=="im-server")].env[?(@.name=="MYSQL_REGISTRY_MAX_IDLE_CONNS")].value}')" + echo "effective requests=${request_cpu}/${request_memory} limits=${limit_cpu}/${limit_memory} liveness=${liveness} mysql=${mysql_max_open}/${mysql_max_idle} registry=${mysql_registry_max_open}/${mysql_registry_max_idle}" if [ "${request_cpu}" != "1" ] || [ "${request_memory}" != "512Mi" ] || \ [ "${limit_cpu}" != "2" ] || [ "${limit_memory}" != "1Gi" ] || \ - [ "${liveness}" != "/livez" ]; then - echo "::error::ArgoCD reverted StatefulSet resources/probes. Apply deploy/argocd/im-server-test/statefulset-performance-patch.yaml to the im-server-test source." + [ "${liveness}" != "/livez" ] || [ "${mysql_max_open}" != "24" ] || \ + [ "${mysql_max_idle}" != "8" ] || [ "${mysql_registry_max_open}" != "4" ] || \ + [ "${mysql_registry_max_idle}" != "2" ]; then + echo "::error::ArgoCD reverted StatefulSet resources/probes/MySQL pools. Apply deploy/argocd/im-server-test/statefulset-performance-patch.yaml to the im-server-test source." exit 1 fi diff --git a/deploy/argocd/im-server-test/README.md b/deploy/argocd/im-server-test/README.md index 47d49d8..eb6835d 100644 --- a/deploy/argocd/im-server-test/README.md +++ b/deploy/argocd/im-server-test/README.md @@ -14,8 +14,12 @@ After ArgoCD sync, verify the effective configuration: ```bash kubectl -n im-server-system get statefulset im-server -o jsonpath='{.spec.template.spec.containers[0].resources}' kubectl -n im-server-system get statefulset im-server -o jsonpath='{.spec.template.spec.containers[0].livenessProbe.httpGet.path}' +kubectl -n im-server-system get statefulset im-server -o jsonpath='{.spec.template.spec.containers[0].env}' ``` Expected values are requests `1 CPU / 512Mi`, limits `2 CPU / 1Gi`, and -liveness path `/livez`. The application repository's deploy workflow waits -for reconciliation and fails if ArgoCD restores different values. +liveness path `/livez`. The MySQL pool environment must be +`MYSQL_MAX_OPEN_CONNS=24`, `MYSQL_MAX_IDLE_CONNS=8`, +`MYSQL_REGISTRY_MAX_OPEN_CONNS=4`, and `MYSQL_REGISTRY_MAX_IDLE_CONNS=2`. +The application repository's deploy workflow waits for reconciliation and +fails if ArgoCD restores different values. diff --git a/deploy/argocd/im-server-test/statefulset-performance-patch.yaml b/deploy/argocd/im-server-test/statefulset-performance-patch.yaml index d0e5693..6237700 100644 --- a/deploy/argocd/im-server-test/statefulset-performance-patch.yaml +++ b/deploy/argocd/im-server-test/statefulset-performance-patch.yaml @@ -19,6 +19,14 @@ spec: value: "3" - name: CLUSTER_ROUTE_RETRY_BACKOFF_MS value: "25" + - name: MYSQL_MAX_OPEN_CONNS + value: "24" + - name: MYSQL_MAX_IDLE_CONNS + value: "8" + - name: MYSQL_REGISTRY_MAX_OPEN_CONNS + value: "4" + - name: MYSQL_REGISTRY_MAX_IDLE_CONNS + value: "2" resources: requests: cpu: "1" diff --git a/docker/tinode/config.template b/docker/tinode/config.template index 53b8754..17ccdc6 100644 --- a/docker/tinode/config.template +++ b/docker/tinode/config.template @@ -79,8 +79,10 @@ "mysql": { "database": "meeting_im", "dsn": "$MYSQL_DSN", - "max_open_conns": 64, - "max_idle_conns": 64, + "max_open_conns": $MYSQL_MAX_OPEN_CONNS, + "max_idle_conns": $MYSQL_MAX_IDLE_CONNS, + "registry_max_open_conns": $MYSQL_REGISTRY_MAX_OPEN_CONNS, + "registry_max_idle_conns": $MYSQL_REGISTRY_MAX_IDLE_CONNS, "conn_max_lifetime": 60, "sql_timeout": 10, "tenant_registry_db": "$TENANT_REGISTRY_DB", diff --git a/docker/tinode/entrypoint.sh b/docker/tinode/entrypoint.sh index ba61fc3..426d3aa 100644 --- a/docker/tinode/entrypoint.sh +++ b/docker/tinode/entrypoint.sh @@ -18,6 +18,10 @@ else : "${CLUSTER_PROXY_TO_MASTER_ENQUEUE_TIMEOUT_MS:=300}" : "${CLUSTER_ROUTE_RETRIES:=3}" : "${CLUSTER_ROUTE_RETRY_BACKOFF_MS:=25}" + : "${MYSQL_MAX_OPEN_CONNS:=24}" + : "${MYSQL_MAX_IDLE_CONNS:=8}" + : "${MYSQL_REGISTRY_MAX_OPEN_CONNS:=4}" + : "${MYSQL_REGISTRY_MAX_IDLE_CONNS:=2}" if [ -z "$CLUSTER_NODES_JSON" ] ; then CLUSTER_NODES_JSON='[{"name":"tinode-0","addr":"tinode-0:12000"},{"name":"tinode-1","addr":"tinode-1:12001"},{"name":"tinode-2","addr":"tinode-2:12002"}]' fi diff --git a/docs/test/IM-单聊压测Pod日志根因分析与改进方案-20260927.md b/docs/test/IM-单聊压测Pod日志根因分析与改进方案-20260927.md index 7d136fb..6419914 100644 --- a/docs/test/IM-单聊压测Pod日志根因分析与改进方案-20260927.md +++ b/docs/test/IM-单聊压测Pod日志根因分析与改进方案-20260927.md @@ -414,3 +414,23 @@ Run ID：`p2p-500-20260927-01`。A/B 各建立 500 条 WebSocket，共 500 对 2. 修正 IM Server 每 Pod 的 `max_open_conns/max_idle_conns`，或在验证 MySQL CPU、内存、线程和 PXC 限制后提高 `max_connections`；必须保留至少 20% 全局余量。 3. 压测中增加 DB pool `OpenConnections/InUse/Idle/WaitCount/WaitDuration`和 MySQL `Threads_connected/Threads_running/Max_used_connections/Connection_errors_max_connections` 指标。 4. 复测要求 Publish ACK 10000/10000、实时丢失和断连为 0、Kafka 60 秒内归零、MySQL 1040 和清理期 queue full 为 0，然后才能进入 1000 对。 + +### 15.1 为什么 Kafka 可以积压，但本轮仍会报错 + +Kafka 积压只能保护“已经成功写入 Kafka”的消息。正常情况下，consumer 落库变慢只会使 lag 上升，消息会在 Kafka 中等待，不应导致 Publish 报错。本轮的 54 条失败发生在 Kafka 之前： + +1. 普通文本消息先同步查询青少年模式设置，但当前规则无论查询结果都允许消息继续进入内容审核。 +2. P2P 发布在写 Kafka 前调用 `CanSendP2P`，旧实现即使 Redis 已命中好友关系，仍会读取 MySQL 用户和好友记录。 +3. 入口校验、Kafka consumer、历史查询和登录审计共用 MySQL 连接预算。连接耗尽后，入口校验直接返回 500，这些消息没有进入 Kafka，因此无法通过“后续慢慢消费”恢复。 + +### 15.2 IM 侧改造 + +2026-09-27 已完成以下修复： + +- 普通消息不再读取青少年模式表；通话的夜间禁用规则仍保留数据库查询。 +- P2P 热路径先查双向黑名单缓存，再使用好友正缓存；缓存未命中或 Redis 异常时才回源 MySQL。新增拉黑会同时清理双向好友正缓存，避免安全状态陈旧。 +- `meeting_im` 每 Pod 连接池设为 `max_open=24/max_idle=8`，`hotgo` 注册表库独立设为 `max_open=4/max_idle=2`。3 个 Pod 的 IM 理论总上限由 384 降为 84；结合 Workbench 30、Manager 20 的上限，整套服务理论上限为 134，在当前 MySQL 170 连接上限下保留 36 条全局余量。 +- 连接池参数改为环境变量，并同时固化到应用仓库及 ArgoCD 真实配置源。发布流水线会在 ArgoCD 回写后校验四个连接池参数。 +- 新增 `im_db_pool_*` 指标，包括 max/open/in-use/idle/wait count/wait duration/关闭计数，并为连接池 85%/95% 使用率配置预警和严重告警。 + +这些修复的目标是先让入口稳定完成缓存校验和 Kafka enqueue，把 MySQL 慢转化为可观测、可恢复的 Kafka lag，而不是 Publish 500。如果复测时 `im_db_pool_wait_total` 持续增长或 Kafka 仍无法在 60 秒内归零，再根据 PXC 的 CPU/内存/线程容量将 MySQL `max_connections` 持久化提高到 300–400，不使用临时 `SET GLOBAL`代替配置源。 diff --git a/monitoring/im-server-alerts.yaml b/monitoring/im-server-alerts.yaml index 2f017ea..46b8e88 100644 --- a/monitoring/im-server-alerts.yaml +++ b/monitoring/im-server-alerts.yaml @@ -49,6 +49,20 @@ spec: summary: "IM WebSocket connections are imbalanced across pods" - name: im-server-persistence rules: + - alert: IMDatabasePoolHigh + expr: im_db_pool_in_use_connections / clamp_min(im_db_pool_max_open_connections, 1) > 0.85 + for: 1m + labels: + severity: warning + annotations: + summary: "IM database pool usage is above 85%" + - alert: IMDatabasePoolCritical + expr: im_db_pool_in_use_connections / clamp_min(im_db_pool_max_open_connections, 1) > 0.95 + for: 30s + labels: + severity: critical + annotations: + summary: "IM database pool usage is above 95%" - alert: IMKafkaPersistDeadLetter expr: sum by (kind, reason) (increase(im_kafka_dead_letters_total[5m])) > 0 labels: diff --git a/server/db/mysql/adapter.go b/server/db/mysql/adapter.go index ea4c14b..60be245 100644 --- a/server/db/mysql/adapter.go +++ b/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() } diff --git a/server/im_security_policy_runtime.go b/server/im_security_policy_runtime.go index 88dbf9e..1fe7880 100644 --- a/server/im_security_policy_runtime.go +++ b/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() } diff --git a/server/im_security_policy_runtime_test.go b/server/im_security_policy_runtime_test.go index 673201b..3b99440 100644 --- a/server/im_security_policy_runtime_test.go +++ b/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) + } +} diff --git a/server/meetingim.conf b/server/meetingim.conf index 9ba1926..350af04 100644 --- a/server/meetingim.conf +++ b/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 } diff --git a/server/metrics/db_pool.go b/server/metrics/db_pool.go new file mode 100644 index 0000000..f7c550e --- /dev/null +++ b/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) +} diff --git a/server/metrics/db_pool_test.go b/server/metrics/db_pool_test.go new file mode 100644 index 0000000..e7aff6a --- /dev/null +++ b/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) + } + } +} diff --git a/server/store/group_message_runtime_test.go b/server/store/group_message_runtime_test.go index 2231d43..417bc0f 100644 --- a/server/store/group_message_runtime_test.go +++ b/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() diff --git a/server/store/store.go b/server/store/store.go index 50f5642..7b353f5 100644 --- a/server/store/store.go +++ b/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 } diff --git a/server/tinode.conf b/server/tinode.conf index 9ba1926..350af04 100644 --- a/server/tinode.conf +++ b/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 } diff --git a/server/user_verification_test.go b/server/user_verification_test.go index 2771132..b43c1b4 100644 --- a/server/user_verification_test.go +++ b/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 }