From d96f661f30dd4e44ea1713e35d129a2b413fb300 Mon Sep 17 00:00:00 2001 From: richjohnrambo Date: Sun, 27 Sep 2026 16:33:05 +0800 Subject: [PATCH] fix: reduce persistence lag and retry audit deadlocks --- docs/sql/IM-后台消息与登录审计SQL.sql | 1 + server/db/mysql/adapter.go | 4 +- server/db/mysql/admin_governance_schema.go | 5 ++ .../db/mysql/admin_governance_schema_test.go | 8 +- server/db/mysql/tenant_security_audit.go | 13 ++- server/db/mysql/tenant_security_audit_test.go | 49 +++++++++++ server/store/message_batch_test.go | 42 ++++++++++ server/store/store.go | 81 ++++++++++++++++--- 8 files changed, 185 insertions(+), 18 deletions(-) create mode 100644 server/store/message_batch_test.go diff --git a/docs/sql/IM-后台消息与登录审计SQL.sql b/docs/sql/IM-后台消息与登录审计SQL.sql index 2f20a71..5457ed0 100644 --- a/docs/sql/IM-后台消息与登录审计SQL.sql +++ b/docs/sql/IM-后台消息与登录审计SQL.sql @@ -26,6 +26,7 @@ CREATE TABLE IF NOT EXISTS im_user_login_logs ( created_at DATETIME(3) NOT NULL DEFAULT CURRENT_TIMESTAMP(3), PRIMARY KEY (id), INDEX idx_im_user_login_tenant_time (tenant_id, login_at DESC, id DESC), + INDEX idx_im_user_login_session_close (tenant_id, session_id, logout_at), INDEX idx_im_user_login_user_time (user_id, login_at DESC, id DESC), INDEX idx_im_user_login_ip_time (ip, login_at DESC, id DESC), INDEX idx_im_user_login_device_time (device_id(64), login_at DESC), diff --git a/server/db/mysql/adapter.go b/server/db/mysql/adapter.go index 7219c6a..ea4c14b 100644 --- a/server/db/mysql/adapter.go +++ b/server/db/mysql/adapter.go @@ -4671,7 +4671,7 @@ func (a *adapter) FileUserHasAccess(fid string, uid t.Uid) (bool, error) { return a.db.GetContext(ctx, &allowed, query, args...) } err := run() - for attempt := 0; isRetryableMySQLReadErr(err) && attempt < 2; attempt++ { + for attempt := 0; isRetryableMySQLError(err) && attempt < 2; attempt++ { time.Sleep(time.Duration(attempt+1) * 25 * time.Millisecond) err = run() } @@ -4813,7 +4813,7 @@ func isIDOutOfRange(err error) bool { return ok && myerr.Number == 1264 && strings.Contains(strings.ToLower(myerr.Message), "id") } -func isRetryableMySQLReadErr(err error) bool { +func isRetryableMySQLError(err error) bool { if err == nil { return false } diff --git a/server/db/mysql/admin_governance_schema.go b/server/db/mysql/admin_governance_schema.go index 8396d96..53f5039 100644 --- a/server/db/mysql/admin_governance_schema.go +++ b/server/db/mysql/admin_governance_schema.go @@ -17,6 +17,7 @@ func ensureIMAdminGovernanceTables(exec sqlExecutor) error { request_id VARCHAR(64) NOT NULL DEFAULT '', login_at DATETIME(3) NOT NULL DEFAULT CURRENT_TIMESTAMP(3), logout_at DATETIME(3), metadata JSON, created_at DATETIME(3) NOT NULL DEFAULT CURRENT_TIMESTAMP(3), PRIMARY KEY(id), INDEX idx_im_user_login_tenant_time(tenant_id, login_at DESC, id DESC), + INDEX idx_im_user_login_session_close(tenant_id, session_id, logout_at), INDEX idx_im_user_login_user_time(user_id, login_at DESC, id DESC), INDEX idx_im_user_login_ip_time(ip, login_at DESC, id DESC), INDEX idx_im_user_login_device_time(device_id(64), login_at DESC), @@ -53,5 +54,9 @@ func ensureIMAdminGovernanceTables(exec sqlExecutor) error { return err } } + if err := execIgnoreMySQLError(exec, + `ALTER TABLE im_user_login_logs ADD INDEX idx_im_user_login_session_close(tenant_id,session_id,logout_at)`, 1061); err != nil { + return err + } return nil } diff --git a/server/db/mysql/admin_governance_schema_test.go b/server/db/mysql/admin_governance_schema_test.go index 948ec86..74a62d6 100644 --- a/server/db/mysql/admin_governance_schema_test.go +++ b/server/db/mysql/admin_governance_schema_test.go @@ -13,8 +13,8 @@ func TestEnsureIMAdminGovernanceTablesDefinesLoginAuditTenantScope(t *testing.T) if err := ensureIMAdminGovernanceTables(exec); err != nil { t.Fatal(err) } - if len(exec.statements) != 3 { - t.Fatalf("created %d admin governance tables, want 3", len(exec.statements)) + if len(exec.statements) != 4 { + t.Fatalf("executed %d admin governance schema statements, want 4", len(exec.statements)) } joined := strings.ToLower(strings.Join(exec.statements, "\n")) for _, table := range []string{"im_user_login_logs", "im_message_admin_actions", "im_user_ip_whitelists"} { @@ -38,4 +38,8 @@ func TestEnsureIMAdminGovernanceTablesDefinesLoginAuditTenantScope(t *testing.T) if end < 0 || !strings.Contains(joined[start:start+end], "tenant_id bigint not null") { t.Fatal("im_user_login_logs must declare tenant_id") } + if !strings.Contains(joined, "idx_im_user_login_session_close(tenant_id, session_id, logout_at)") && + !strings.Contains(joined, "idx_im_user_login_session_close(tenant_id,session_id,logout_at)") { + t.Fatal("im_user_login_logs must index tenant-scoped session close updates") + } } diff --git a/server/db/mysql/tenant_security_audit.go b/server/db/mysql/tenant_security_audit.go index 4481f3c..0dc203c 100644 --- a/server/db/mysql/tenant_security_audit.go +++ b/server/db/mysql/tenant_security_audit.go @@ -93,7 +93,18 @@ func (a *adapter) TenantLoginAuditClose(id t.TenantID, sessionID string, at time if tenantAdp != a { return tenantAdp.TenantLoginAuditClose(id, sessionID, at) } - _, err = a.db.Exec("UPDATE im_user_login_logs SET logout_at=? WHERE tenant_id=? AND session_id=? AND logout_at IS NULL", at, int64(id), sessionID) + return retryMySQLDeadlock(func() error { + _, execErr := a.db.Exec("UPDATE im_user_login_logs SET logout_at=? WHERE tenant_id=? AND session_id=? AND logout_at IS NULL", at, int64(id), sessionID) + return execErr + }, time.Sleep) +} + +func retryMySQLDeadlock(operation func() error, sleep func(time.Duration)) error { + err := operation() + for attempt := 1; isRetryableMySQLError(err) && attempt < 4; attempt++ { + sleep(time.Duration(attempt) * 25 * time.Millisecond) + err = operation() + } return err } diff --git a/server/db/mysql/tenant_security_audit_test.go b/server/db/mysql/tenant_security_audit_test.go index 46550a5..3a0e714 100644 --- a/server/db/mysql/tenant_security_audit_test.go +++ b/server/db/mysql/tenant_security_audit_test.go @@ -3,9 +3,13 @@ package mysql import ( + "errors" "os" "strings" "testing" + "time" + + mysqldriver "github.com/go-sql-driver/mysql" ) func TestLoginAuditQueriesAreTenantScoped(t *testing.T) { @@ -24,3 +28,48 @@ func TestLoginAuditQueriesAreTenantScoped(t *testing.T) { } } } + +func TestRetryMySQLDeadlockEventuallySucceeds(t *testing.T) { + calls := 0 + var delays []time.Duration + err := retryMySQLDeadlock(func() error { + calls++ + if calls < 3 { + return &mysqldriver.MySQLError{Number: 1213, SQLState: [5]byte{'4', '0', '0', '0', '1'}} + } + return nil + }, func(delay time.Duration) { delays = append(delays, delay) }) + if err != nil { + t.Fatal(err) + } + if calls != 3 { + t.Fatalf("operation calls=%d want=3", calls) + } + want := []time.Duration{25 * time.Millisecond, 50 * time.Millisecond} + if len(delays) != len(want) || delays[0] != want[0] || delays[1] != want[1] { + t.Fatalf("retry delays=%v want=%v", delays, want) + } +} + +func TestRetryMySQLDeadlockStopsOnNonRetryableError(t *testing.T) { + want := errors.New("invalid update") + calls := 0 + err := retryMySQLDeadlock(func() error { + calls++ + return want + }, func(time.Duration) { t.Fatal("non-retryable error must not sleep") }) + if !errors.Is(err, want) || calls != 1 { + t.Fatalf("err=%v calls=%d want=%v/1", err, calls, want) + } +} + +func TestRetryMySQLDeadlockStopsAfterFourAttempts(t *testing.T) { + calls := 0 + err := retryMySQLDeadlock(func() error { + calls++ + return &mysqldriver.MySQLError{Number: 1205} + }, func(time.Duration) {}) + if err == nil || calls != 4 { + t.Fatalf("err=%v calls=%d want non-nil/4", err, calls) + } +} diff --git a/server/store/message_batch_test.go b/server/store/message_batch_test.go new file mode 100644 index 0000000..729ea6a --- /dev/null +++ b/server/store/message_batch_test.go @@ -0,0 +1,42 @@ +package store + +import ( + "testing" + + "github.com/tinode/chat/server/store/types" +) + +func TestAggregatePreparedMessageBatchUpdatesKeepsLatestSeq(t *testing.T) { + user1 := types.Uid(1) + user2 := types.Uid(2) + items := []PreparedMessageBatchItem{ + {Message: &types.Message{Topic: "p2p-b", From: user1.String(), SeqId: 3}, ReadBySender: true}, + {Message: &types.Message{Topic: "p2p-a", From: user1.String(), SeqId: 2}, ReadBySender: true}, + {Message: &types.Message{Topic: "p2p-a", From: user2.String(), SeqId: 5}, ReadBySender: true}, + {Message: &types.Message{Topic: "p2p-a", From: user1.String(), SeqId: 6}, ReadBySender: true}, + {Message: &types.Message{Topic: "p2p-b", From: user1.String(), SeqId: 7}, ReadBySender: false}, + {}, + } + + topics, reads := aggregatePreparedMessageBatchUpdates(items) + if len(topics) != 2 || topics[0].topic != "p2p-a" || topics[0].message.SeqId != 6 || + topics[1].topic != "p2p-b" || topics[1].message.SeqId != 7 { + t.Fatalf("unexpected topic updates: %+v", topics) + } + if len(reads) != 3 { + t.Fatalf("read updates=%d want=3: %+v", len(reads), reads) + } + got := make(map[string]int64, len(reads)) + for _, update := range reads { + got[update.topic+":"+update.from.String()] = update.message.SeqId + } + for key, want := range map[string]int64{ + "p2p-a:" + user1.String(): 6, + "p2p-a:" + user2.String(): 5, + "p2p-b:" + user1.String(): 3, + } { + if got[key] != want { + t.Fatalf("read update %s=%d want=%d; all=%v", key, got[key], want, got) + } + } +} diff --git a/server/store/store.go b/server/store/store.go index 5283c57..50f5642 100644 --- a/server/store/store.go +++ b/server/store/store.go @@ -3682,6 +3682,58 @@ type PreparedMessageBatchItem struct { ReadBySender bool } +type preparedMessageBatchTopicUpdate struct { + topic string + message *types.Message +} + +type preparedMessageBatchReadUpdate struct { + topic string + from types.Uid + message *types.Message +} + +func aggregatePreparedMessageBatchUpdates(items []PreparedMessageBatchItem) ([]preparedMessageBatchTopicUpdate, []preparedMessageBatchReadUpdate) { + topics := make(map[string]*types.Message) + reads := make(map[string]preparedMessageBatchReadUpdate) + for _, item := range items { + msg := item.Message + if msg == nil { + continue + } + if previous := topics[msg.Topic]; previous == nil || msg.SeqId > previous.SeqId { + topics[msg.Topic] = msg + } + from := types.ParseUid(msg.From) + if !item.ReadBySender || from.IsZero() { + continue + } + key := msg.Topic + "\x00" + from.String() + if previous, ok := reads[key]; !ok || msg.SeqId > previous.message.SeqId { + reads[key] = preparedMessageBatchReadUpdate{topic: msg.Topic, from: from, message: msg} + } + } + topicKeys := make([]string, 0, len(topics)) + for topic := range topics { + topicKeys = append(topicKeys, topic) + } + sort.Strings(topicKeys) + topicUpdates := make([]preparedMessageBatchTopicUpdate, 0, len(topicKeys)) + for _, topic := range topicKeys { + topicUpdates = append(topicUpdates, preparedMessageBatchTopicUpdate{topic: topic, message: topics[topic]}) + } + readKeys := make([]string, 0, len(reads)) + for key := range reads { + readKeys = append(readKeys, key) + } + sort.Strings(readKeys) + readUpdates := make([]preparedMessageBatchReadUpdate, 0, len(readKeys)) + for _, key := range readKeys { + readUpdates = append(readUpdates, reads[key]) + } + return topicUpdates, readUpdates +} + func PersistPreparedMessageBatch(tenantID types.TenantID, items []PreparedMessageBatchItem) error { if len(items) == 0 { return nil @@ -3725,14 +3777,17 @@ func PersistPreparedMessageBatch(tenantID types.TenantID, items []PreparedMessag if err != nil { return err } - for _, item := range items { - msg := item.Message - from := types.ParseUid(msg.From) - if err = tenantAdp.TenantTopicUpdateOnMessage(tenantID, msg.Topic, msg); err != nil { + topicUpdates, readUpdates := aggregatePreparedMessageBatchUpdates(items) + for _, update := range topicUpdates { + if err = tenantAdp.TenantTopicUpdateOnMessage(tenantID, update.topic, update.message); err != nil { return err } - invalidateTopicCache(tenantID, msg.Topic) - if isGroup && groupMsgRuntime != nil { + invalidateTopicCache(tenantID, update.topic) + } + if isGroup && groupMsgRuntime != nil { + for _, item := range items { + msg := item.Message + from := types.ParseUid(msg.From) if msg.ClientMsgID != "" { _ = groupMsgRuntime.MarkDone(tenantID, msg.Topic, from, msg.ClientMsgID, msg.SeqId, msg.Uid().String()) } @@ -3741,13 +3796,13 @@ func PersistPreparedMessageBatch(tenantID types.TenantID, items []PreparedMessag }) _ = groupMsgRuntime.CacheHotMessage(tenantID, msg) } - if item.ReadBySender && !from.IsZero() { - if updateErr := tenantAdp.TenantSubsUpdate(tenantID, msg.Topic, from, map[string]any{ - "RecvSeqId": msg.SeqId, "ReadSeqId": msg.SeqId, - }); updateErr == nil { - invalidateSubscriptionCache(tenantID, msg.Topic, from) - DeleteCachedUserUnreadCount(tenantID, from) - } + } + for _, update := range readUpdates { + if updateErr := tenantAdp.TenantSubsUpdate(tenantID, update.topic, update.from, map[string]any{ + "RecvSeqId": update.message.SeqId, "ReadSeqId": update.message.SeqId, + }); updateErr == nil { + invalidateSubscriptionCache(tenantID, update.topic, update.from) + DeleteCachedUserUnreadCount(tenantID, update.from) } } return nil