From 7b25365ec3e43373cffa09b0d2c5a151e83eacf1 Mon Sep 17 00:00:00 2001 From: richjohnrambo Date: Sun, 4 Oct 2026 00:49:27 +0800 Subject: [PATCH] fix(group): bound push pressure and synchronize load agents --- docker/tinode/Dockerfile | 2 + docker/tinode/Dockerfile.build | 2 + docker/tinode/config.template | 4 +- docker/tinode/entrypoint.sh | 1 + docs/IM-006商户双机消息压力测试方案.md | 2 +- ...up2k-r9-双机群聊压测分析与改进-20261003.md | 73 +++++++ loadtest/im-pressure/README.md | 2 +- loadtest/im-pressure/agent.go | 3 + loadtest/im-pressure/coordinator.go | 101 +++++++++- loadtest/im-pressure/coordinator_test.go | 50 +++++ server/meetingim.conf | 4 +- server/push/jpush/push_jpush.go | 185 ++++++++++++++++-- server/push/jpush/push_jpush_test.go | 90 +++++++-- server/tinode.conf | 4 +- 14 files changed, 476 insertions(+), 47 deletions(-) create mode 100644 docs/test/IM-group2k-r9-双机群聊压测分析与改进-20261003.md diff --git a/docker/tinode/Dockerfile b/docker/tinode/Dockerfile index 91dbe14..880c78c 100644 --- a/docker/tinode/Dockerfile +++ b/docker/tinode/Dockerfile @@ -194,6 +194,8 @@ ENV TNPG_ORG= ENV JPUSH_PUSH_ENABLED=false ENV JPUSH_API_DOMAIN=https://api.jpush.cn ENV JPUSH_PUSH_PATH=/v3/push +ENV JPUSH_WORKERS=8 +ENV JPUSH_DEVICE_CACHE_TTL_SECONDS=30 ENV JPUSH_DEFAULT_TTL_SECONDS=86400 ENV JPUSH_CALL_TTL_SECONDS=60 ENV JPUSH_APNS_PRODUCTION=false diff --git a/docker/tinode/Dockerfile.build b/docker/tinode/Dockerfile.build index efa50f2..e6dd4ff 100644 --- a/docker/tinode/Dockerfile.build +++ b/docker/tinode/Dockerfile.build @@ -110,6 +110,8 @@ ENV TNPG_PUSH_ENABLED=false ENV JPUSH_PUSH_ENABLED=false ENV JPUSH_API_DOMAIN=https://api.jpush.cn ENV JPUSH_PUSH_PATH=/v3/push +ENV JPUSH_WORKERS=8 +ENV JPUSH_DEVICE_CACHE_TTL_SECONDS=30 ENV JPUSH_DEFAULT_TTL_SECONDS=86400 ENV JPUSH_CALL_TTL_SECONDS=60 ENV JPUSH_APNS_PRODUCTION=false diff --git a/docker/tinode/config.template b/docker/tinode/config.template index 53a1025..7811d6a 100644 --- a/docker/tinode/config.template +++ b/docker/tinode/config.template @@ -85,7 +85,7 @@ "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, + "conn_max_lifetime": $MYSQL_CONN_MAX_LIFETIME_SECONDS, "sql_timeout": 10, "tenant_registry_db": "$TENANT_REGISTRY_DB", "tenant_db_routing": true @@ -266,6 +266,8 @@ "enabled": $JPUSH_PUSH_ENABLED, "api_domain": "$JPUSH_API_DOMAIN", "push_path": "$JPUSH_PUSH_PATH", + "workers": $JPUSH_WORKERS, + "device_cache_ttl_seconds": $JPUSH_DEVICE_CACHE_TTL_SECONDS, "default_ttl_seconds": $JPUSH_DEFAULT_TTL_SECONDS, "call_ttl_seconds": $JPUSH_CALL_TTL_SECONDS, "apns_production": $JPUSH_APNS_PRODUCTION, diff --git a/docker/tinode/entrypoint.sh b/docker/tinode/entrypoint.sh index c57d7d3..7001c67 100644 --- a/docker/tinode/entrypoint.sh +++ b/docker/tinode/entrypoint.sh @@ -24,6 +24,7 @@ else : "${MYSQL_MAX_IDLE_CONNS:=8}" : "${MYSQL_REGISTRY_MAX_OPEN_CONNS:=4}" : "${MYSQL_REGISTRY_MAX_IDLE_CONNS:=2}" + : "${MYSQL_CONN_MAX_LIFETIME_SECONDS:=1800}" 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/IM-006商户双机消息压力测试方案.md b/docs/IM-006商户双机消息压力测试方案.md index f462e90..b74f871 100644 --- a/docs/IM-006商户双机消息压力测试方案.md +++ b/docs/IM-006商户双机消息压力测试方案.md @@ -737,7 +737,7 @@ loadtest/im-pressure/internal/collector/ loadtest/im-pressure/internal/report/ ``` -协调方式：Machine A 启动 coordinator，等待 A/B 两个 agent 都上报 1000 个 READY，然后发布一个至少晚于当前时间 30 秒的 UTC `start_at`。两台机器按相同 `run_id` 和确定性调度执行。任何 agent 失联都将本轮标记为 `INVALID`。 +协调方式：Machine A 启动 coordinator，等待 A/B 两个 agent 都上报 1000 个 READY，然后发布一个至少晚于当前时间 30 秒的 UTC `start_at`。到达 `start_at` 后，A/B 必须在前后 3 秒内向 coordinator 完成最终起跑确认；只有双方都确认后才开始发送，任一方迟到则两边共同终止并将本轮标记为 `INVALID`。两台机器按相同 `run_id` 和确定性调度执行，任何 agent 失联同样判定为 `INVALID`。 `agent` 自身暴露 `/metrics`，提供连接状态、发布、接收、重复、乱序、断连及 Go/process 指标；ACK 和投递延迟分布写入最终 summary。`collect` 只执行只读查询，并把各数据源的原始快照与查询时间一起保存；连接凭证通过环境变量传入，不写入结果目录。 diff --git a/docs/test/IM-group2k-r9-双机群聊压测分析与改进-20261003.md b/docs/test/IM-group2k-r9-双机群聊压测分析与改进-20261003.md new file mode 100644 index 0000000..0c333d9 --- /dev/null +++ b/docs/test/IM-group2k-r9-双机群聊压测分析与改进-20261003.md @@ -0,0 +1,73 @@ +# GROUP-2000 R9 双机群聊压测分析与改进 + +## 1. 结论 + +`group2k-r9-20261003-01` 不能作为 2000 人群聊容量结论。消息最终落库完整，但实时阶段同时发生机器 B 迟到、Pod 重启和集群 DNS 拒绝查询，导致大量连接断开和跨机器投递缺失。 + +- 发送尝试 2000 条，数据库精确找到 1774 条服务端已确认消息，缺失、重复、字段不一致均为 0。 +- 实时期望投递 3,548,000 次，实际收到 1,384,872 次，到达率约 39.03%。 +- 机器 B 比 coordinator 的计划起跑时间晚约 7 分 55 秒；B 未收到任何 A 消息，本轮并非有效的同步双机负载。 +- `im-server-1` 在负载中退出，容器退出码 137；现有权限无法读取 Event 和 Pod metrics，尚不能把原因唯一归为 OOM、liveness 或节点压力。 +- 三个 Pod 合计约 1715 次依赖域名解析失败，日志包含对集群 DNS `10.43.0.10:53` 的 `connection refused`。 +- JPush 设备查询产生 792 次数据库错误；用户更新队列满 1132 次，属于依赖抖动和连接重建后的次生压力。 + +Kafka 持久化链路本轮表现正确：各 Pod 批量持久化计数合计 1774，与数据库精确核验一致，之前的失败消息保留修复已生效。 + +## 2. 根因链路 + +### 2.1 JPush 查询按群消息放大 + +JPush 使用 registration ID 模式时，每条群消息都会按全部收件人重新查询 `im_device_push_tokens`。2000 人群和 2000 条消息会形成约 400 万个用户查询输入；测试账号没有推送设备，旧实现仍重复查询空结果。JPush 回执又以无限 goroutine 并发处理，使查询、对象分配和内存压力同时放大。 + +### 2.2 MySQL 连接每 60 秒回收 + +旧配置的 `conn_max_lifetime=60` 会在持续压测期间频繁重建连接。每次建连都重新解析 `mysql.118im.com`；当集群 DNS 短时拒绝请求时，业务查询、JPush 查询和 Kafka 持久化会一起失败或等待。 + +### 2.3 双机只有计划时间，没有最终到达屏障 + +旧协议在双方 READY 后下发统一时间，但 agent 到达该时间后直接发送。机器 B 晚启动时，机器 A 无法知道对方没有真正进入发送阶段，因此形成两段错开的单机流量并生成无效的跨机器统计。 + +## 3. 已完成改进 + +### 3.1 JPush + +- 设备查询结果按租户、provider、用户缓存 30 秒，包含“无设备”结果。 +- 并发首次查询被合并，同一批群成员只访问一次数据库。 +- 查询失败后退避 1 秒，已有缓存可在 5 分钟陈旧窗口内继续使用。 +- 缓存定期清理，避免用户集合长期增长。 +- JPush 回执限制为 8 个工作协程，取消无限 goroutine，队列负责形成背压。 + +该改动不删除在线多端静默推送语义。新注册或更新的设备最迟约 30 秒进入推送查询结果。 + +### 3.2 MySQL 与 DNS + +- 容器默认 `conn_max_lifetime` 从 60 秒提高到 1800 秒，并支持 `MYSQL_CONN_MAX_LIFETIME_SECONDS` 覆盖。 +- 测试环境 GitOps 显式设置 1800 秒，降低持续负载中的连接重建和 DNS 查询频率。 +- 保留现有 `ndots:1`、`timeout:2`、`attempts:2`。 + +这会显著减少应用制造的 DNS 压力，但不能替代集群 DNS 修复。若复测仍出现 `10.43.0.10:53 connection refused`，需要集群管理员检查 CoreDNS 副本、CPU、转发上游、conntrack，并评估与当前 K3s 网络模式匹配的 NodeLocal DNSCache；不能在未知网络模式下直接套用 DaemonSet。 + +### 3.3 压测同步 + +- coordinator 增加最终起跑屏障。 +- A/B 必须在计划时间前后 3 秒内到达，双方都到达后才开始发送。 +- 任一机器迟到时，两边收到同一失败原因并终止本轮。 +- coordinator 输出保存双方实际到达时间和 `start_error`。 + +### 3.4 诊断权限 + +`backend-debug` 只读权限增加 Event 和 `metrics.k8s.io/pods`，用于下一轮区分 OOM、探针失败、驱逐和节点资源压力。若集群未安装 metrics-server，Pod metrics 仍不可用，但 Event 读取不受影响。 + +## 4. 复测门禁 + +下一轮使用新 `run_id`，A/B 必须更新到包含起跑屏障的同一提交，并在应用和 GitOps 配置都发布后执行。 + +1. coordinator 输出中 A/B `started` 时间相差不超过 3 秒，`start_error` 为空。 +2. 2000 条发送全部得到明确 ACK；数据库按 `client_mid` 精确匹配，缺失、重复、字段不一致均为 0。 +3. 实时期望 4,000,000 次投递全部到达，重复、乱序和断连均为 0。 +4. JPush 设备查询不再随群消息数线性增长，日志中 `jpush: db error` 为 0。 +5. Pod 不重启，Event 中无 OOM、Unhealthy、Evicted；负载窗口内持续采集 CPU、内存和重启次数。 +6. `10.43.0.10:53`、MySQL、Redis、Kafka 域名解析错误均为 0。 +7. Kafka lag 最终归零，数据库最终消息数与 ACK 台账一致。 + +若业务正确性通过但 DNS 仍有拒绝请求，本轮仍判基础设施门禁失败，不继续用增加应用重试掩盖 DNS 容量问题。 diff --git a/loadtest/im-pressure/README.md b/loadtest/im-pressure/README.md index ded1ac9..cf01add 100644 --- a/loadtest/im-pressure/README.md +++ b/loadtest/im-pressure/README.md @@ -46,7 +46,7 @@ - 两台机器必须使用 NTP/chrony 同步时间。报告会分别输出同机时钟延迟和跨机器单向延迟；跨机器出现负延迟样本时，后者仅用于诊断，不参与门禁。 - agent 默认每 20 秒发送 WebSocket Ping，覆盖等待另一台机器、发压、drain 和历史核验阶段；`ws_ping_interval` 必须小于公网入口的空闲连接超时。 - 正式开始前 agent 会持续检查实际存活连接；连接断开后自动重新登录并订阅 P2P、群 Topic。初次连接不齐或等待另一台机器期间掉线不会退出。 -- coordinator 使用两阶段屏障：两边实时连接数都达到配置值后进入验证倒计时；倒计时内任一方掉线就撤销屏障。两边再次确认满连接后才提交最终开始时间。 +- coordinator 使用三阶段屏障：两边实时连接数都达到配置值后进入验证倒计时；倒计时内任一方掉线就撤销屏障；提交最终开始时间后，两边必须在该时间前后 3 秒内再次到达起跑屏障，且双方都到达后才允许发送。任一方迟到会让两边共同终止本轮，避免生成错时压测结果。 - 压测机如保持 OpenVPN 等全局 TUN，仅在另一套代理中设置 `DIRECT` 无法绕过它。起压前必须确认 `http_base` 和 WS 域名的每个当前 IP 都走物理网卡；需保留 VPN 访问内网时，可为这些 IP 配置经局域网关的 `/32` 主机路由。VPN 重连、DNS 变更或机器重启后需重新核验。 ## 统计口径 diff --git a/loadtest/im-pressure/agent.go b/loadtest/im-pressure/agent.go index 9e3e09d..e51e36e 100644 --- a/loadtest/im-pressure/agent.go +++ b/loadtest/im-pressure/agent.go @@ -186,6 +186,9 @@ func runAgent(ctx context.Context, cfg *Config, manifest *Manifest, machine, coo if err = waitUntil(ctx, start.StartAt); err != nil { return err } + if err = coord.started(ctx, machine, start.StartAt); err != nil { + return fmt.Errorf("coordinated start rejected: %w", err) + } if alive := run.aliveConnections(); alive != len(accounts) { return fmt.Errorf("connections alive at scheduled start=%d expected=%d", alive, len(accounts)) } diff --git a/loadtest/im-pressure/coordinator.go b/loadtest/im-pressure/coordinator.go index c2db3ea..0fd8678 100644 --- a/loadtest/im-pressure/coordinator.go +++ b/loadtest/im-pressure/coordinator.go @@ -3,6 +3,7 @@ package main import ( "context" "encoding/json" + "errors" "fmt" "io" "log" @@ -29,6 +30,20 @@ type AgentReady struct { Armed bool `json:"armed"` } +type AgentStarted struct { + Machine string `json:"machine"` + ScheduledAt time.Time `json:"scheduled_at"` +} + +type StartGate struct { + Proceed bool `json:"proceed"` + Error string `json:"error,omitempty"` + ArrivedAgents int `json:"arrived_agents"` + ExpectedAgents int `json:"expected_agents"` +} + +const maxAgentStartSkew = 3 * time.Second + type coordinatorState struct { mu sync.Mutex start StartInfo @@ -37,6 +52,8 @@ type coordinatorState struct { expectedAgents int ready map[string]AgentReady armed map[string]bool + started map[string]time.Time + startError string ledgers map[string][]AcceptedMessage observed map[string][]AcceptedMessage history map[string][]string @@ -60,11 +77,13 @@ func runCoordinator(ctx context.Context, listen, runID, scenario, output string, expectedAgents: expectedAgents, ready: make(map[string]AgentReady), ledgers: make(map[string][]AcceptedMessage), observed: make(map[string][]AcceptedMessage), history: make(map[string][]string), - armed: make(map[string]bool), summaries: make(map[string]AgentSummary), complete: make(chan struct{}), + armed: make(map[string]bool), started: make(map[string]time.Time), + summaries: make(map[string]AgentSummary), complete: make(chan struct{}), } mux := http.NewServeMux() mux.HandleFunc("/v1/ready", state.handleReady) mux.HandleFunc("/v1/start", state.handleStart) + mux.HandleFunc("/v1/started", state.handleStarted) mux.HandleFunc("/v1/ledger", state.handleLedger) mux.HandleFunc("/v1/observed", state.handleObserved) mux.HandleFunc("/v1/history", state.handleHistory) @@ -210,11 +229,15 @@ func (s *coordinatorState) handleReady(w http.ResponseWriter, r *http.Request) { s.start.StartAt = time.Time{} s.start.Committed = false clear(s.armed) + clear(s.started) + s.startError = "" } if allStable && s.start.StartAt.IsZero() { s.start.StartAt = now.Add(s.startDelay) s.start.Committed = false clear(s.armed) + clear(s.started) + s.startError = "" log.Printf("all agents ready; validation_at=%s", s.start.StartAt.Format(time.RFC3339Nano)) } if allStable && !s.start.Committed && ready.Armed && !now.Before(s.start.StartAt) { @@ -234,6 +257,43 @@ func (s *coordinatorState) handleReady(w http.ResponseWriter, r *http.Request) { writeJSON(w, http.StatusOK, start) } +func (s *coordinatorState) handleStarted(w http.ResponseWriter, r *http.Request) { + if r.Method != http.MethodPost { + http.Error(w, "method not allowed", http.StatusMethodNotAllowed) + return + } + var arrival AgentStarted + if !decodeLimitedJSON(w, r, &arrival) || (arrival.Machine != "a" && arrival.Machine != "b") || arrival.ScheduledAt.IsZero() { + http.Error(w, "invalid start arrival", http.StatusBadRequest) + return + } + + s.mu.Lock() + defer s.mu.Unlock() + if !s.start.Committed || !arrival.ScheduledAt.Equal(s.start.StartAt) { + writeJSON(w, http.StatusOK, StartGate{Error: "start schedule changed", ExpectedAgents: s.expectedAgents}) + return + } + if s.started == nil { + s.started = make(map[string]time.Time) + } + if _, exists := s.started[arrival.Machine]; !exists { + now := time.Now().UTC() + s.started[arrival.Machine] = now + skew := now.Sub(s.start.StartAt) + if skew < -maxAgentStartSkew || skew > maxAgentStartSkew { + s.startError = fmt.Sprintf("agent %s reached start barrier %s from schedule (limit %s)", + arrival.Machine, skew.Round(time.Millisecond), maxAgentStartSkew) + log.Printf("start barrier rejected: %s", s.startError) + } + } + gate := StartGate{ + Proceed: len(s.started) == s.expectedAgents && s.startError == "", Error: s.startError, + ArrivedAgents: len(s.started), ExpectedAgents: s.expectedAgents, + } + writeJSON(w, http.StatusOK, gate) +} + func (s *coordinatorState) handleStart(w http.ResponseWriter, r *http.Request) { s.mu.Lock() defer s.mu.Unlock() @@ -311,7 +371,8 @@ func (s *coordinatorState) handleStatus(w http.ResponseWriter, _ *http.Request) s.mu.Lock() defer s.mu.Unlock() writeJSON(w, http.StatusOK, map[string]any{ - "start": s.start, "ready": s.ready, "armed": s.armed, + "start": s.start, "ready": s.ready, "armed": s.armed, "started": s.started, + "start_error": s.startError, "ledger_agents": len(s.ledgers), "observed_agents": len(s.observed), "history_agents": len(s.history), "summary_agents": len(s.summaries), }) @@ -320,14 +381,17 @@ func (s *coordinatorState) handleStatus(w http.ResponseWriter, _ *http.Request) func (s *coordinatorState) writeOutput(path string) error { s.mu.Lock() data := struct { - Start StartInfo `json:"start"` - Ready map[string]AgentReady `json:"ready"` - Ledgers map[string][]AcceptedMessage `json:"ledgers"` - Observed map[string][]AcceptedMessage `json:"observed"` - History map[string][]string `json:"history"` - Summaries map[string]AgentSummary `json:"summaries"` + Start StartInfo `json:"start"` + Ready map[string]AgentReady `json:"ready"` + Started map[string]time.Time `json:"started"` + StartError string `json:"start_error,omitempty"` + Ledgers map[string][]AcceptedMessage `json:"ledgers"` + Observed map[string][]AcceptedMessage `json:"observed"` + History map[string][]string `json:"history"` + Summaries map[string]AgentSummary `json:"summaries"` }{ - Start: s.start, Ready: s.ready, Ledgers: s.ledgers, Observed: s.observed, + Start: s.start, Ready: s.ready, Started: s.started, StartError: s.startError, + Ledgers: s.ledgers, Observed: s.observed, History: s.history, Summaries: s.summaries, } s.mu.Unlock() @@ -403,6 +467,25 @@ func (c *coordinatorClient) submitLedger(ctx context.Context, machine string, me return err } +func (c *coordinatorClient) started(ctx context.Context, machine string, scheduledAt time.Time) error { + arrival := AgentStarted{Machine: machine, ScheduledAt: scheduledAt} + for { + var gate StartGate + if _, err := c.request(ctx, http.MethodPost, "/v1/started", arrival, &gate); err != nil { + return err + } + if gate.Error != "" { + return errors.New(gate.Error) + } + if gate.Proceed { + return nil + } + if err := waitContext(ctx, 100*time.Millisecond); err != nil { + return err + } + } +} + func (c *coordinatorClient) ledger(ctx context.Context) ([]AcceptedMessage, error) { for { var messages []AcceptedMessage diff --git a/loadtest/im-pressure/coordinator_test.go b/loadtest/im-pressure/coordinator_test.go index 57fecd1..c183b15 100644 --- a/loadtest/im-pressure/coordinator_test.go +++ b/loadtest/im-pressure/coordinator_test.go @@ -15,6 +15,7 @@ func TestCoordinatorResetsBarrierWhenConnectionCountDrops(t *testing.T) { start: StartInfo{RunID: "run-reset", Scenario: "group"}, startDelay: time.Minute, commitDelay: time.Second, expectedAgents: 2, ready: make(map[string]AgentReady), armed: make(map[string]bool), ledgers: make(map[string][]AcceptedMessage), + started: make(map[string]time.Time), summaries: make(map[string]AgentSummary), complete: make(chan struct{}), } postReady := func(ready AgentReady) StartInfo { @@ -52,6 +53,7 @@ func TestCoordinatorClientWaitsForBothAgentsAndMergesLedger(t *testing.T) { commitDelay: 50 * time.Millisecond, expectedAgents: 2, ready: make(map[string]AgentReady), ledgers: make(map[string][]AcceptedMessage), armed: make(map[string]bool), summaries: make(map[string]AgentSummary), complete: make(chan struct{}), + started: make(map[string]time.Time), } server := httptest.NewServer(testCoordinatorMux(state)) defer server.Close() @@ -116,6 +118,7 @@ func testCoordinatorMux(state *coordinatorState) *http.ServeMux { mux := http.NewServeMux() mux.HandleFunc("/v1/ready", state.handleReady) mux.HandleFunc("/v1/start", state.handleStart) + mux.HandleFunc("/v1/started", state.handleStarted) mux.HandleFunc("/v1/ledger", state.handleLedger) mux.HandleFunc("/v1/observed", state.handleObserved) mux.HandleFunc("/v1/history", state.handleHistory) @@ -123,3 +126,50 @@ func testCoordinatorMux(state *coordinatorState) *http.ServeMux { mux.HandleFunc("/v1/status", state.handleStatus) return mux } + +func TestCoordinatorRejectsLateAgentForBothMachines(t *testing.T) { + startAt := time.Now().Add(-maxAgentStartSkew - time.Second).UTC() + state := &coordinatorState{ + start: StartInfo{RunID: "run-late", Scenario: "group", StartAt: startAt, Committed: true}, + expectedAgents: 2, started: make(map[string]time.Time), + } + server := httptest.NewServer(testCoordinatorMux(state)) + defer server.Close() + client := newCoordinatorClient(server.URL) + if err := client.started(context.Background(), "b", startAt); err == nil { + t.Fatal("late agent must be rejected") + } + if err := client.started(context.Background(), "a", startAt); err == nil { + t.Fatal("the other agent must observe the same rejected run") + } +} + +func TestCoordinatorReleasesAgentsOnlyAfterBothReachStart(t *testing.T) { + startAt := time.Now().UTC() + state := &coordinatorState{ + start: StartInfo{RunID: "run-synced", Scenario: "group", StartAt: startAt, Committed: true}, + expectedAgents: 2, started: make(map[string]time.Time), + } + server := httptest.NewServer(testCoordinatorMux(state)) + defer server.Close() + client := newCoordinatorClient(server.URL) + aDone := make(chan error, 1) + go func() { aDone <- client.started(context.Background(), "a", startAt) }() + time.Sleep(50 * time.Millisecond) + select { + case err := <-aDone: + t.Fatalf("machine A passed before B arrived: %v", err) + default: + } + if err := client.started(context.Background(), "b", startAt); err != nil { + t.Fatalf("machine B start: %v", err) + } + select { + case err := <-aDone: + if err != nil { + t.Fatalf("machine A start: %v", err) + } + case <-time.After(time.Second): + t.Fatal("machine A was not released") + } +} diff --git a/server/meetingim.conf b/server/meetingim.conf index 3357731..b5f0b64 100644 --- a/server/meetingim.conf +++ b/server/meetingim.conf @@ -89,7 +89,7 @@ "max_idle_conns": 8, "registry_max_open_conns": 4, "registry_max_idle_conns": 2, - "conn_max_lifetime": 60, + "conn_max_lifetime": 1800, "sql_timeout": 10 } }, @@ -229,6 +229,8 @@ "enabled": false, "api_domain": "https://api.jpush.cn", "push_path": "/v3/push", + "workers": 8, + "device_cache_ttl_seconds": 30, "default_ttl_seconds": 86400, "call_ttl_seconds": 60, "apns_production": false, diff --git a/server/push/jpush/push_jpush.go b/server/push/jpush/push_jpush.go index 81f5233..86cd3be 100644 --- a/server/push/jpush/push_jpush.go +++ b/server/push/jpush/push_jpush.go @@ -12,6 +12,7 @@ import ( "os" "strconv" "strings" + "sync" "time" "github.com/tinode/chat/server/logs" @@ -21,16 +22,39 @@ import ( ) const ( - defaultBufferSize = 1024 - defaultAPIDomain = "https://api.jpush.cn" - defaultPushPath = "/v3/push" - defaultTTL = 86400 - defaultCallTTL = 30 - callCategory = "im_call" + defaultBufferSize = 1024 + defaultWorkers = 8 + defaultAPIDomain = "https://api.jpush.cn" + defaultPushPath = "/v3/push" + defaultTTL = 86400 + defaultCallTTL = 30 + defaultDeviceCacheTTL = 30 + callCategory = "im_call" ) var handler Handler +type deviceCacheKey struct { + tenantID t.TenantID + provider string + uid t.Uid +} + +type deviceCacheEntry struct { + devices []t.DeviceDef + expiresAt time.Time + staleUntil time.Time +} + +type providerDeviceCache struct { + mu sync.Mutex + entries map[deviceCacheKey]deviceCacheEntry + retryAfter time.Time + nextCleanup time.Time +} + +var jpushDeviceCache = providerDeviceCache{entries: make(map[deviceCacheKey]deviceCacheEntry)} + type Handler struct { input chan *push.Receipt channel chan *push.ChannelReq @@ -40,19 +64,21 @@ type Handler struct { } type configType struct { - Enabled bool `json:"enabled"` - DryRun bool `json:"dry_run"` - AppKey string `json:"app_key"` - MasterSecret string `json:"master_secret"` - APIDomain string `json:"api_domain"` - PushPath string `json:"push_path"` - Buffer int `json:"buffer"` - DefaultTTLSeconds int `json:"default_ttl_seconds"` - CallTTLSeconds int `json:"call_ttl_seconds"` - APNsProduction bool `json:"apns_production"` - UseAlias *bool `json:"use_alias"` - Classification int `json:"classification"` - Title string `json:"title"` + Enabled bool `json:"enabled"` + DryRun bool `json:"dry_run"` + AppKey string `json:"app_key"` + MasterSecret string `json:"master_secret"` + APIDomain string `json:"api_domain"` + PushPath string `json:"push_path"` + Buffer int `json:"buffer"` + Workers int `json:"workers"` + DefaultTTLSeconds int `json:"default_ttl_seconds"` + CallTTLSeconds int `json:"call_ttl_seconds"` + DeviceCacheTTLSeconds int `json:"device_cache_ttl_seconds"` + APNsProduction bool `json:"apns_production"` + UseAlias *bool `json:"use_alias"` + Classification int `json:"classification"` + Title string `json:"title"` } type jpushPayload struct { @@ -154,10 +180,19 @@ func (Handler) Init(jsonconf json.RawMessage) (bool, error) { handler.config = normalized go func() { + workers := make(chan struct{}, normalized.Workers) for { select { case rcpt := <-handler.input: - go sendReceipt(rcpt, &handler.config, handler.client) + select { + case workers <- struct{}{}: + go func(receipt *push.Receipt) { + defer func() { <-workers }() + sendReceipt(receipt, &handler.config, handler.client) + }(rcpt) + case <-handler.stop: + return + } case req := <-handler.channel: logs.Info.Println("jpush: channel subscription ignored", req.Uid.UserId(), req.Channel) case <-handler.stop: @@ -186,12 +221,18 @@ func normalizeConfig(config configType) (configType, error) { if config.Buffer <= 0 { config.Buffer = defaultBufferSize } + if config.Workers <= 0 { + config.Workers = defaultWorkers + } if config.DefaultTTLSeconds <= 0 { config.DefaultTTLSeconds = defaultTTL } if config.CallTTLSeconds <= 0 { config.CallTTLSeconds = defaultCallTTL } + if config.DeviceCacheTTLSeconds <= 0 { + config.DeviceCacheTTLSeconds = defaultDeviceCacheTTL + } if config.Title == "" { config.Title = "MeetingIM" } @@ -260,6 +301,8 @@ func loadThirdPartyConfig(config *configType) { "jpush_master_secret", "jpush_api_domain", "jpush_push_path", + "jpush_workers", + "jpush_device_cache_ttl_seconds", "jpush_default_ttl_seconds", "jpush_call_ttl_seconds", "jpush_apns_production", @@ -285,6 +328,12 @@ func loadThirdPartyConfig(config *configType) { if v := strings.TrimSpace(values["jpush_push_path"]); v != "" { config.PushPath = v } + if v, ok := parseConfigInt(values["jpush_workers"]); ok { + config.Workers = v + } + if v, ok := parseConfigInt(values["jpush_device_cache_ttl_seconds"]); ok { + config.DeviceCacheTTLSeconds = v + } if v, ok := parseConfigInt(values["jpush_default_ttl_seconds"]); ok { config.DefaultTTLSeconds = v } @@ -344,7 +393,8 @@ func preparePayloads(rcpt *push.Receipt, config *configType) ([]*jpushPayload, [ } var devices map[t.Uid][]t.DeviceDef if !useAlias { - devices, _, err = providerDevicesForUsers(rcpt.TenantID, "jpush", uids...) + devices, _, err = cachedProviderDevicesForUsers(rcpt.TenantID, "jpush", + time.Duration(config.DeviceCacheTTLSeconds)*time.Second, uids...) if err != nil { logs.Warn.Println("jpush: db error", err) return nil, nil @@ -699,6 +749,99 @@ func providerDevicesForUsers(tenantID t.TenantID, provider string, uids ...t.Uid return map[t.Uid][]t.DeviceDef{}, 0, nil } +func cachedProviderDevicesForUsers(tenantID t.TenantID, provider string, ttl time.Duration, + uids ...t.Uid) (map[t.Uid][]t.DeviceDef, int, error) { + if ttl <= 0 { + ttl = defaultDeviceCacheTTL * time.Second + } + provider = strings.ToLower(strings.TrimSpace(provider)) + now := time.Now() + cache := &jpushDeviceCache + cache.mu.Lock() + defer cache.mu.Unlock() + + result := make(map[t.Uid][]t.DeviceDef, len(uids)) + misses := make([]t.Uid, 0, len(uids)) + for _, uid := range uids { + key := deviceCacheKey{tenantID: tenantID, provider: provider, uid: uid} + entry, ok := cache.entries[key] + if ok && now.Before(entry.expiresAt) { + result[uid] = append([]t.DeviceDef(nil), entry.devices...) + continue + } + misses = append(misses, uid) + } + if len(misses) == 0 { + return result, countDevices(result), nil + } + if cache.nextCleanup.IsZero() || !now.Before(cache.nextCleanup) { + for key, entry := range cache.entries { + if !now.Before(entry.staleUntil) { + delete(cache.entries, key) + } + } + cache.nextCleanup = now.Add(time.Minute) + } + + // A failed lookup used to trigger one identical database query per group + // message. Back off briefly and use stale entries where possible instead. + if now.Before(cache.retryAfter) { + cache.addStale(result, tenantID, provider, now, misses) + return result, countDevices(result), nil + } + + loaded, _, err := providerDevicesForUsers(tenantID, provider, misses...) + if err != nil { + cache.retryAfter = now.Add(time.Second) + if cache.addStale(result, tenantID, provider, now, misses) == len(misses) { + return result, countDevices(result), nil + } + return result, countDevices(result), err + } + cache.retryAfter = time.Time{} + staleTTL := 10 * ttl + if staleTTL < 5*time.Minute { + staleTTL = 5 * time.Minute + } + for _, uid := range misses { + devices := append([]t.DeviceDef(nil), loaded[uid]...) + cache.entries[deviceCacheKey{tenantID: tenantID, provider: provider, uid: uid}] = deviceCacheEntry{ + devices: devices, expiresAt: now.Add(ttl), staleUntil: now.Add(staleTTL), + } + result[uid] = append([]t.DeviceDef(nil), devices...) + } + return result, countDevices(result), nil +} + +func (c *providerDeviceCache) addStale(result map[t.Uid][]t.DeviceDef, tenantID t.TenantID, + provider string, now time.Time, uids []t.Uid) int { + covered := 0 + for _, uid := range uids { + entry, ok := c.entries[deviceCacheKey{tenantID: tenantID, provider: provider, uid: uid}] + if ok && now.Before(entry.staleUntil) { + result[uid] = append([]t.DeviceDef(nil), entry.devices...) + covered++ + } + } + return covered +} + +func countDevices(devices map[t.Uid][]t.DeviceDef) int { + count := 0 + for _, entries := range devices { + count += len(entries) + } + return count +} + +func resetDeviceCache() { + jpushDeviceCache.mu.Lock() + jpushDeviceCache.entries = make(map[deviceCacheKey]deviceCacheEntry) + jpushDeviceCache.retryAfter = time.Time{} + jpushDeviceCache.nextCleanup = time.Time{} + jpushDeviceCache.mu.Unlock() +} + func (Handler) IsReady() bool { return handler.input != nil } diff --git a/server/push/jpush/push_jpush_test.go b/server/push/jpush/push_jpush_test.go index 815f511..90e2b18 100644 --- a/server/push/jpush/push_jpush_test.go +++ b/server/push/jpush/push_jpush_test.go @@ -5,6 +5,7 @@ import ( "encoding/json" "net/http" "net/http/httptest" + "sync" "testing" "time" @@ -27,32 +28,39 @@ func (s testThirdPartyConfigStore) Get(_ string, keys ...string) (map[string]str type testDevicePushTokenStore struct { devices map[types.Uid][]types.DeviceDef + calls int } func (s *testDevicePushTokenStore) Upsert(types.TenantID, *types.DevicePushToken) (*types.DevicePushToken, error) { return nil, nil } -func (s *testDevicePushTokenStore) GetAllByProvider(types.TenantID, string, ...types.Uid) (map[types.Uid][]types.DeviceDef, int, error) { +func (s *testDevicePushTokenStore) GetAllByProvider(_ types.TenantID, _ string, uids ...types.Uid) (map[types.Uid][]types.DeviceDef, int, error) { + s.calls++ + result := make(map[types.Uid][]types.DeviceDef, len(uids)) count := 0 - for _, devices := range s.devices { + for _, uid := range uids { + devices := s.devices[uid] + result[uid] = devices count += len(devices) } - return s.devices, count, nil + return result, count, nil } func TestNormalizeConfigLoadsThirdPartyConfig(t *testing.T) { oldConfig := store.ThirdPartyConfig store.ThirdPartyConfig = testThirdPartyConfigStore{ - "jpush_enabled": "1", - "jpush_app_key": "app-key", - "jpush_master_secret": "master", - "jpush_api_domain": "https://push.example.com/", - "jpush_push_path": "v3/push", - "jpush_default_ttl_seconds": "123", - "jpush_call_ttl_seconds": "45", - "jpush_use_alias": "false", - "jpush_classification": "1", + "jpush_enabled": "1", + "jpush_app_key": "app-key", + "jpush_master_secret": "master", + "jpush_api_domain": "https://push.example.com/", + "jpush_push_path": "v3/push", + "jpush_workers": "4", + "jpush_device_cache_ttl_seconds": "45", + "jpush_default_ttl_seconds": "123", + "jpush_call_ttl_seconds": "45", + "jpush_use_alias": "false", + "jpush_classification": "1", } defer func() { store.ThirdPartyConfig = oldConfig }() @@ -68,6 +76,9 @@ func TestNormalizeConfigLoadsThirdPartyConfig(t *testing.T) { if config.APIDomain != "https://push.example.com" || config.PushPath != "/v3/push" { t.Fatalf("unexpected endpoint: %#v", config) } + if config.Workers != 4 || config.DeviceCacheTTLSeconds != 45 { + t.Fatalf("unexpected worker/cache config: %#v", config) + } if config.DefaultTTLSeconds != 123 || config.CallTTLSeconds != 45 || config.UseAlias == nil || *config.UseAlias { t.Fatalf("unexpected ttl/use_alias: %#v", config) } @@ -119,6 +130,8 @@ func TestAliasForUidUsesJPushSafeCharacters(t *testing.T) { } func TestPreparePayloadsCanUseRegistrationID(t *testing.T) { + resetDeviceCache() + t.Cleanup(resetDeviceCache) oldTokens := store.DevicePushTokens uid := types.Uid(42) store.DevicePushTokens = &testDevicePushTokenStore{devices: map[types.Uid][]types.DeviceDef{ @@ -147,6 +160,8 @@ func TestPreparePayloadsCanUseRegistrationID(t *testing.T) { } func TestPreparePayloadsDoesNotDropRecipientWithDeliveredSession(t *testing.T) { + resetDeviceCache() + t.Cleanup(resetDeviceCache) oldTokens := store.DevicePushTokens uid := types.Uid(42) store.DevicePushTokens = &testDevicePushTokenStore{devices: map[types.Uid][]types.DeviceDef{ @@ -170,6 +185,57 @@ func TestPreparePayloadsDoesNotDropRecipientWithDeliveredSession(t *testing.T) { } } +func TestPreparePayloadsCachesEmptyDeviceLookups(t *testing.T) { + resetDeviceCache() + t.Cleanup(resetDeviceCache) + oldTokens := store.DevicePushTokens + tokens := &testDevicePushTokenStore{devices: map[types.Uid][]types.DeviceDef{}} + store.DevicePushTokens = tokens + t.Cleanup(func() { store.DevicePushTokens = oldTokens }) + useAlias := false + config := &configType{Title: "MeetingIM", DefaultTTLSeconds: 86400, CallTTLSeconds: 60, + DeviceCacheTTLSeconds: 30, UseAlias: &useAlias} + receipt := &push.Receipt{ + TenantID: 7, + To: map[types.Uid]push.Recipient{42: {}, 43: {}}, + Payload: push.Payload{What: push.ActMsg, Topic: "grpTopic", SeqId: 9, Timestamp: time.Now()}, + } + preparePayloads(receipt, config) + preparePayloads(receipt, config) + if tokens.calls != 1 { + t.Fatalf("device store calls=%d want=1", tokens.calls) + } +} + +func TestPreparePayloadsCoalescesConcurrentDeviceLookups(t *testing.T) { + resetDeviceCache() + t.Cleanup(resetDeviceCache) + oldTokens := store.DevicePushTokens + tokens := &testDevicePushTokenStore{devices: map[types.Uid][]types.DeviceDef{}} + store.DevicePushTokens = tokens + t.Cleanup(func() { store.DevicePushTokens = oldTokens }) + useAlias := false + config := &configType{Title: "MeetingIM", DefaultTTLSeconds: 86400, CallTTLSeconds: 60, + DeviceCacheTTLSeconds: 30, UseAlias: &useAlias} + receipt := &push.Receipt{ + TenantID: 7, + To: map[types.Uid]push.Recipient{42: {}, 43: {}}, + Payload: push.Payload{What: push.ActMsg, Topic: "grpTopic", SeqId: 9, Timestamp: time.Now()}, + } + var wg sync.WaitGroup + for range 20 { + wg.Add(1) + go func() { + defer wg.Done() + preparePayloads(receipt, config) + }() + } + wg.Wait() + if tokens.calls != 1 { + t.Fatalf("device store calls=%d want=1", tokens.calls) + } +} + func TestBuildPayloadUsesShortTTLForCalls(t *testing.T) { expiresAt := time.Now().Add(30 * time.Second).UTC().Format(time.RFC3339Nano) config := &configType{Title: "MeetingIM", DefaultTTLSeconds: 86400, CallTTLSeconds: 60} diff --git a/server/tinode.conf b/server/tinode.conf index eb2bba2..d257e2b 100644 --- a/server/tinode.conf +++ b/server/tinode.conf @@ -88,7 +88,7 @@ "max_idle_conns": 8, "registry_max_open_conns": 4, "registry_max_idle_conns": 2, - "conn_max_lifetime": 60, + "conn_max_lifetime": 1800, "sql_timeout": 10 } }, @@ -228,6 +228,8 @@ "enabled": false, "api_domain": "https://api.jpush.cn", "push_path": "/v3/push", + "workers": 8, + "device_cache_ttl_seconds": 30, "default_ttl_seconds": 86400, "call_ttl_seconds": 60, "apns_production": false,