From 929f847003d5bece09d38de659379193341bb57b Mon Sep 17 00:00:00 2001 From: richjohnrambo Date: Tue, 1 Sep 2026 22:20:33 +0800 Subject: [PATCH] Refactor Kafka publishing without DB outbox --- docs/IM-Kafka接入方案.md | 20 +- docs/IM-Kafka接入联调测试文档.md | 15 +- docs/sql/IM-Kafka-Outbox删除SQL.sql | 4 + docs/sql/IM-Kafka基础表SQL.sql | 23 --- server/db/adapter.go | 4 - server/db/mysql/infrastructure_schema.go | 14 -- server/db/mysql/infrastructure_schema_test.go | 8 +- server/db/mysql/tenant_infrastructure.go | 134 ------------ server/im_http_test.go | 28 +-- server/kafka_admin_commands.go | 24 +-- server/kafka_dead_letters.go | 47 +++++ server/kafka_file_lifecycle.go | 16 +- server/kafka_message_persist.go | 45 ++-- server/kafka_outbox.go | 126 ------------ server/kafka_runtime.go | 5 +- server/kafka_workbench_audit.go | 17 +- server/main.go | 2 +- server/meetingim.conf | 4 +- server/store/group_message_runtime.go | 36 +++- server/store/infrastructure.go | 36 ---- server/store/store.go | 4 + server/store/types/infrastructure.go | 24 --- server/tinode.conf | 4 +- server/topic.go | 61 +++++- server/topic_test.go | 193 ++++++++++++++++++ 25 files changed, 416 insertions(+), 478 deletions(-) create mode 100644 docs/sql/IM-Kafka-Outbox删除SQL.sql delete mode 100644 server/kafka_outbox.go diff --git a/docs/IM-Kafka接入方案.md b/docs/IM-Kafka接入方案.md index c9dfa26..b57668f 100644 --- a/docs/IM-Kafka接入方案.md +++ b/docs/IM-Kafka接入方案.md @@ -722,14 +722,14 @@ IM Server 已接入 `im.message.persist`，用于把在线发送链路和最终 当前处理方式： -1. WS/HTTP 发送入口先同步完成权限校验、业务幂等、`seq_id` 分配和 Topic 游标占用。 -2. Kafka 开启时，发送入口先写租户库 `im_kafka_outbox`，由 Outbox worker 可靠生产 `im.message.persist`，客户端仍立即拿到已分配的 `seq_id`。 +1. WS/HTTP 发送入口先完成权限校验，并由 topic 内存态分配 `seq_id`。 +2. Kafka 开启时，发送入口直接生产 `im.message.persist`，不查库、不落库、不写 DB outbox；客户端在 Kafka 写入成功后拿到已分配的 `seq_id`。 3. Kafka 关闭时，继续走原同步落库路径，兼容本地和未接 Kafka 环境。 4. 消费者按 `message_kind` 区分私聊与群聊： - `p2p` 写入 `im_messages`。 - `group` 写入 `im_group_messages`。 5. 消费侧复用 `im_kafka_consume_log` 做 `event_id + consumer_group` 幂等。 -6. 失败按统一退避重试，超过上限写入 `im_kafka_dead_letters` 并投递 `im.dead.letter`。 +6. 失败按统一退避重试 3 次，超过上限先投递 `im.dead.letter`，再由 dead letter consumer 写入 `im_kafka_dead_letters`。 ### 11.10 IM Server 已接入的缓存失效消费 @@ -874,17 +874,9 @@ IM Server 已在 `im.admin.command` 中接入用户治理类后台动作。Admin } ``` -### 11.9 IM Server 已接入的 Kafka Outbox Worker +### 11.9 IM Server 已移除 Kafka DB Outbox -IM Server 已接入租户库 `im_kafka_outbox` 通用发送 worker，用于避免“业务库写成功但 Kafka 未发送”的事件丢失。 - -当前能力： - -1. 启动后按 active 租户扫描每个租户库的 `im_kafka_outbox`。 -2. 扫描 `pending/failed` 且已到 `next_retry_at` 的事件。 -3. 发送成功后标记 `sent`、写入 `sent_at`。 -4. 发送失败后标记 `failed`，按 Kafka 配置的 `retry_backoff_ms` 写入下一次重试时间。 -5. 首个接入方为 `im.message.persist`，后续文件生命周期、工作台事件、审计事件可复用同一 outbox。 +IM Server 不再维护租户库 `im_kafka_outbox`，Kafka 事件统一由发送入口直接生产。生产侧使用 Kafka writer 的 `max_retry_count` 控制发送重试；消费侧使用 `im_kafka_consume_log` 做处理幂等和失败重试，超过上限后进入 `im.dead.letter`。 本轮暂不实现 `im.message.push` 离线 Push Kafka 化，已记录到 [IM 第一至十一章后续增强项](./IM-未实现后续增强项.md)。 @@ -930,7 +922,7 @@ IM Server 已接入 `im.workbench.event` 的首批生产与消费。 ### 11.12 IM Server 已接入的审计事件 -IM Server 已接入 `im.audit.event` 的可靠生产，审计事件写入租户库 `im_kafka_outbox` 后由 Kafka Outbox worker 发送。 +IM Server 已接入 `im.audit.event` 的直接生产，审计事件不写 DB outbox。 当前审计范围： diff --git a/docs/IM-Kafka接入联调测试文档.md b/docs/IM-Kafka接入联调测试文档.md index 5fa0a54..76e750b 100644 --- a/docs/IM-Kafka接入联调测试文档.md +++ b/docs/IM-Kafka接入联调测试文档.md @@ -54,7 +54,6 @@ SOURCE docs/sql/IM-Kafka基础表SQL.sql; | 表 | 用途 | | --- | --- | | `im_kafka_consume_log` | 消费幂等、重试状态。 | -| `im_kafka_outbox` | IM 侧可靠生产 Kafka 事件。 | | `im_kafka_dead_letters` | 死信记录与人工处理状态。 | | `im_messages` | 私聊消息存储。 | | `im_group_messages` | 群聊消息存储。 | @@ -886,15 +885,7 @@ GET /v1/internal/admin-query/tenants/default/kafka/dead-letters/{dead_id} ### TC-KAFKA-030 Outbox 可靠投递 -目标：验证 IM 侧 Kafka Outbox 防止业务成功但 Kafka 丢事件。 - -步骤： - -1. 临时停止 Kafka。 -2. 执行会产生 outbox 的操作，如发送消息、提交工作台事件。 -3. 查询 `im_kafka_outbox`。 -4. 恢复 Kafka。 -5. 等待 outbox worker 投递。 +目标：验证 IM 侧 Kafka 直接生产失败时不会产生 DB outbox，调用方收到发送失败。 预期： @@ -958,7 +949,7 @@ GET /v1/internal/admin-query/tenants/default/kafka/dead-letters/{dead_id} | 进入死信 | 查 `im_kafka_dead_letters.envelope/last_error`，修复后用重放命令。 | | 用户没有被踢下线 | 查目标 `user_id` 是否是 IM 用户 ID；是否连接在当前 IM 节点；跨节点踢人是否需要后续 Redis fanout 扩展。 | | 群治理未生效 | 查 `topic/version/actor_user_id` 是否正确，目标用户是否为群成员。 | -| 消息未落库 | 查 `im_kafka_outbox` 是否 sent，`im.message.persist` 是否消费成功。 | +| 消息未落库 | 查 `im.message.persist` 是否消费成功，以及 `im_kafka_consume_log.last_reason/last_error`。 | | 附件未绑定 | 查消息是否带附件、`im.file.lifecycle` 是否消费成功。 | | 内部 API 401 | 查 secret、签名文本、RequestURI、timestamp、nonce 是否一致。 | | 配置没有刷新 | 查是否发送 `im.config.event` 或 `im.cache.invalidate`，IM 是否清理了对应 namespace。 | @@ -1023,7 +1014,7 @@ GET /v1/internal/admin-query/tenants/default/kafka/dead-letters/{dead_id} | 文件生命周期 | 上传完成、附件绑定、群文件登记/删除、OSS 元数据和引用表。 | | 工作台事件 | 会议、日报/月报、总经理信箱、红点/系统通知投递。 | | 死信闭环 | 死信查询、忽略、解决、重放，以及重放后二次成功/失败。 | -| Outbox | Kafka 停止/恢复时 `im_kafka_outbox`、Admin outbox 的可靠投递。 | +| Kafka 生产失败 | Kafka 停止/恢复时，确认 IM 不写 `im_kafka_outbox`，接口/WS 返回发送失败，恢复后可重新发送。 | | Admin 页面 | 操作单状态从 pending/running 到 succeeded/failed/dead 的页面联调。 | ### 10.2 2026-08-27 完整后端链路联调 diff --git a/docs/sql/IM-Kafka-Outbox删除SQL.sql b/docs/sql/IM-Kafka-Outbox删除SQL.sql new file mode 100644 index 0000000..2f090cc --- /dev/null +++ b/docs/sql/IM-Kafka-Outbox删除SQL.sql @@ -0,0 +1,4 @@ +-- Remove the deprecated Kafka DB outbox table from a tenant database. +-- Run once per tenant business database after deploying the direct Kafka publish path. + +DROP TABLE IF EXISTS im_kafka_outbox; diff --git a/docs/sql/IM-Kafka基础表SQL.sql b/docs/sql/IM-Kafka基础表SQL.sql index 9d4cae6..1f91efd 100644 --- a/docs/sql/IM-Kafka基础表SQL.sql +++ b/docs/sql/IM-Kafka基础表SQL.sql @@ -29,29 +29,6 @@ CREATE TABLE IF NOT EXISTS im_kafka_consume_log ( INDEX idx_im_kafka_consume_log_topic(source_topic, id) ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_0900_ai_ci; -CREATE TABLE IF NOT EXISTS im_kafka_outbox ( - id BIGINT NOT NULL AUTO_INCREMENT, - event_id VARCHAR(96) NOT NULL, - topic VARCHAR(128) NOT NULL, - event_type VARCHAR(96) NOT NULL, - biz_type VARCHAR(64) NOT NULL DEFAULT '', - biz_id VARCHAR(128) NOT NULL DEFAULT '', - key_value VARCHAR(256) NOT NULL DEFAULT '', - envelope JSON NOT NULL, - status VARCHAR(16) NOT NULL DEFAULT 'pending', - attempts INT NOT NULL DEFAULT 0, - next_retry_at DATETIME(3), - sent_at DATETIME(3), - last_reason VARCHAR(64) NOT NULL DEFAULT '', - last_error VARCHAR(1000) NOT NULL DEFAULT '', - created_at DATETIME(3) NOT NULL DEFAULT CURRENT_TIMESTAMP(3), - updated_at DATETIME(3) NOT NULL DEFAULT CURRENT_TIMESTAMP(3) ON UPDATE CURRENT_TIMESTAMP(3), - PRIMARY KEY(id), - UNIQUE INDEX uk_im_kafka_outbox_event(event_id), - INDEX idx_im_kafka_outbox_pending(status, next_retry_at, id), - INDEX idx_im_kafka_outbox_biz(biz_type, biz_id) -) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_0900_ai_ci; - CREATE TABLE IF NOT EXISTS im_kafka_dead_letters ( id BIGINT NOT NULL AUTO_INCREMENT, dead_id VARCHAR(96) NOT NULL, diff --git a/server/db/adapter.go b/server/db/adapter.go index 8558d6f..2886328 100644 --- a/server/db/adapter.go +++ b/server/db/adapter.go @@ -263,10 +263,6 @@ type TenantHTTPInfrastructureAdapter interface { TenantKafkaDeadLetterGet(t.TenantID, string) (*t.KafkaDeadLetter, error) TenantKafkaDeadLetterMark(t.TenantID, string, string, string, string, time.Time) error TenantKafkaDeadLetterMarkReplay(t.TenantID, string, string, string, time.Time) (*t.KafkaDeadLetter, error) - TenantKafkaOutboxCreate(t.TenantID, *t.KafkaOutboxRecord) error - TenantKafkaOutboxListPending(t.TenantID, int, time.Time) ([]t.KafkaOutboxRecord, error) - TenantKafkaOutboxMarkSent(t.TenantID, string, time.Time) error - TenantKafkaOutboxMarkFailed(t.TenantID, string, string, string, time.Time, time.Time) error } // TenantRegistrationInviteAdapter stores invitation definitions in the diff --git a/server/db/mysql/infrastructure_schema.go b/server/db/mysql/infrastructure_schema.go index a789b6c..179ca9d 100644 --- a/server/db/mysql/infrastructure_schema.go +++ b/server/db/mysql/infrastructure_schema.go @@ -44,20 +44,6 @@ func ensureIMHTTPInfrastructureTables(exec sqlExecutor) error { INDEX idx_im_kafka_consume_log_biz(biz_type, biz_id), INDEX idx_im_kafka_consume_log_topic(source_topic, id) ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_0900_ai_ci`, - `CREATE TABLE IF NOT EXISTS im_kafka_outbox ( - id BIGINT NOT NULL AUTO_INCREMENT, event_id VARCHAR(96) NOT NULL, - topic VARCHAR(128) NOT NULL, event_type VARCHAR(96) NOT NULL, - biz_type VARCHAR(64) NOT NULL DEFAULT '', biz_id VARCHAR(128) NOT NULL DEFAULT '', - key_value VARCHAR(256) NOT NULL DEFAULT '', envelope JSON NOT NULL, - status VARCHAR(16) NOT NULL DEFAULT 'pending', attempts INT NOT NULL DEFAULT 0, - next_retry_at DATETIME(3), sent_at DATETIME(3), last_reason VARCHAR(64) NOT NULL DEFAULT '', - last_error VARCHAR(1000) NOT NULL DEFAULT '', - created_at DATETIME(3) NOT NULL DEFAULT CURRENT_TIMESTAMP(3), - updated_at DATETIME(3) NOT NULL DEFAULT CURRENT_TIMESTAMP(3) ON UPDATE CURRENT_TIMESTAMP(3), - PRIMARY KEY(id), UNIQUE INDEX uk_im_kafka_outbox_event(event_id), - INDEX idx_im_kafka_outbox_pending(status, next_retry_at, id), - INDEX idx_im_kafka_outbox_biz(biz_type, biz_id) - ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_0900_ai_ci`, `CREATE TABLE IF NOT EXISTS im_kafka_dead_letters ( id BIGINT NOT NULL AUTO_INCREMENT, dead_id VARCHAR(96) NOT NULL, event_id VARCHAR(96) NOT NULL, parent_event_id VARCHAR(96) NOT NULL DEFAULT '', diff --git a/server/db/mysql/infrastructure_schema_test.go b/server/db/mysql/infrastructure_schema_test.go index 34435e3..e8d8f24 100644 --- a/server/db/mysql/infrastructure_schema_test.go +++ b/server/db/mysql/infrastructure_schema_test.go @@ -12,15 +12,14 @@ func TestEnsureIMHTTPInfrastructureTablesUsesSelectedTenantDatabase(t *testing.T if err := ensureIMHTTPInfrastructureTables(exec); err != nil { t.Fatal(err) } - if len(exec.statements) != 8 { - t.Fatalf("created %d infrastructure tables, want 8", len(exec.statements)) + if len(exec.statements) != 7 { + t.Fatalf("created %d infrastructure tables, want 7", len(exec.statements)) } joined := strings.ToLower(strings.Join(exec.statements, "\n")) for _, table := range []string{ "im_http_idempotency", "im_business_events_outbox", "im_kafka_consume_log", - "im_kafka_outbox", "im_kafka_dead_letters", "im_realname_appeals", "im_conversation_settings", @@ -33,4 +32,7 @@ func TestEnsureIMHTTPInfrastructureTablesUsesSelectedTenantDatabase(t *testing.T if strings.Contains(joined, "tinode.") || strings.Contains(joined, "create database") || strings.Contains(joined, "tenant_id") { t.Fatalf("infrastructure schema must stay inside the selected tenant database: %s", joined) } + if strings.Contains(joined, "im_kafka_outbox") { + t.Fatalf("kafka outbox table must not be created: %s", joined) + } } diff --git a/server/db/mysql/tenant_infrastructure.go b/server/db/mysql/tenant_infrastructure.go index e9e6c8c..b11092e 100644 --- a/server/db/mysql/tenant_infrastructure.go +++ b/server/db/mysql/tenant_infrastructure.go @@ -512,140 +512,6 @@ func (a *adapter) TenantKafkaDeadLetterMarkReplay(id t.TenantID, deadID, resolve return a.TenantKafkaDeadLetterGet(id, deadID) } -func (a *adapter) TenantKafkaOutboxCreate(id t.TenantID, record *t.KafkaOutboxRecord) error { - tenantAdp, err := a.infrastructureAdapter(id) - if err != nil { - return err - } - if record == nil || record.EventID == "" || record.Topic == "" || record.EventType == "" || len(record.Envelope) == 0 { - return t.ErrMalformed - } - if tenantAdp != a { - return tenantAdp.TenantKafkaOutboxCreate(id, record) - } - now := t.TimeNow() - _, err = a.db.Exec(`INSERT INTO im_kafka_outbox - (event_id,topic,event_type,biz_type,biz_id,key_value,envelope,status,next_retry_at,created_at,updated_at) - VALUES(?,?,?,?,?,?,?,?,?,?,?) - ON DUPLICATE KEY UPDATE updated_at=updated_at`, - record.EventID, record.Topic, record.EventType, record.BizType, record.BizID, record.Key, - record.Envelope, t.KafkaOutboxPending, record.NextRetryAt, now, now) - return err -} - -func (a *adapter) TenantKafkaOutboxListPending(id t.TenantID, limit int, now time.Time) ([]t.KafkaOutboxRecord, error) { - tenantAdp, err := a.infrastructureAdapter(id) - if err != nil { - return nil, err - } - if limit <= 0 { - limit = 100 - } - if tenantAdp != a { - return tenantAdp.TenantKafkaOutboxListPending(id, limit, now) - } - rows := []struct { - ID int64 `db:"id"` - EventID string `db:"event_id"` - Topic string `db:"topic"` - EventType string `db:"event_type"` - BizType string `db:"biz_type"` - BizID string `db:"biz_id"` - Key string `db:"key_value"` - Envelope []byte `db:"envelope"` - Status string `db:"status"` - Attempts int `db:"attempts"` - NextRetryAt sql.NullTime `db:"next_retry_at"` - SentAt sql.NullTime `db:"sent_at"` - LastReason string `db:"last_reason"` - LastError string `db:"last_error"` - CreatedAt time.Time `db:"created_at"` - UpdatedAt time.Time `db:"updated_at"` - }{} - err = a.db.Select(&rows, `SELECT id,event_id,topic,event_type,biz_type,biz_id,key_value,envelope,status,attempts, - next_retry_at,sent_at,last_reason,last_error,created_at,updated_at - FROM im_kafka_outbox - WHERE status IN (?,?) AND (next_retry_at IS NULL OR next_retry_at<=?) - ORDER BY id ASC LIMIT ?`, t.KafkaOutboxPending, t.KafkaOutboxFailed, now, limit) - if err != nil { - return nil, err - } - out := make([]t.KafkaOutboxRecord, 0, len(rows)) - for _, row := range rows { - rec := t.KafkaOutboxRecord{ - ID: row.ID, - EventID: row.EventID, - Topic: row.Topic, - EventType: row.EventType, - BizType: row.BizType, - BizID: row.BizID, - Key: row.Key, - Envelope: row.Envelope, - Status: row.Status, - Attempts: row.Attempts, - LastReason: row.LastReason, - LastError: row.LastError, - CreatedAt: row.CreatedAt, - UpdatedAt: row.UpdatedAt, - } - if row.NextRetryAt.Valid { - rec.NextRetryAt = &row.NextRetryAt.Time - } - if row.SentAt.Valid { - rec.SentAt = &row.SentAt.Time - } - out = append(out, rec) - } - return out, nil -} - -func (a *adapter) TenantKafkaOutboxMarkSent(id t.TenantID, eventID string, now time.Time) error { - tenantAdp, err := a.infrastructureAdapter(id) - if err != nil { - return err - } - if eventID == "" { - return t.ErrMalformed - } - if tenantAdp != a { - return tenantAdp.TenantKafkaOutboxMarkSent(id, eventID, now) - } - res, err := a.db.Exec(`UPDATE im_kafka_outbox SET status=?,sent_at=?,attempts=attempts+1,last_reason='', - last_error='',updated_at=? WHERE event_id=? AND status<>?`, - t.KafkaOutboxSent, now, now, eventID, t.KafkaOutboxSent) - if err != nil { - return err - } - if affected, _ := res.RowsAffected(); affected == 0 { - return t.ErrNotFound - } - return nil -} - -func (a *adapter) TenantKafkaOutboxMarkFailed(id t.TenantID, eventID, reason, message string, nextRetryAt, now time.Time) error { - tenantAdp, err := a.infrastructureAdapter(id) - if err != nil { - return err - } - if eventID == "" { - return t.ErrMalformed - } - if tenantAdp != a { - return tenantAdp.TenantKafkaOutboxMarkFailed(id, eventID, reason, message, nextRetryAt, now) - } - res, err := a.db.Exec(`UPDATE im_kafka_outbox SET status=?,attempts=attempts+1,next_retry_at=?, - last_reason=?,last_error=?,updated_at=? WHERE event_id=? AND status<>?`, - t.KafkaOutboxFailed, nextRetryAt, truncateKafkaString(reason, 64), truncateKafkaString(message, 1000), - now, eventID, t.KafkaOutboxSent) - if err != nil { - return err - } - if affected, _ := res.RowsAffected(); affected == 0 { - return t.ErrNotFound - } - return nil -} - func scanKafkaDeadLetter(rows *sqlx.Rows) (*t.KafkaDeadLetter, error) { var row struct { ID int64 `db:"id"` diff --git a/server/im_http_test.go b/server/im_http_test.go index 3da08ff..45258ae 100644 --- a/server/im_http_test.go +++ b/server/im_http_test.go @@ -38,9 +38,9 @@ type testVerificationStore struct { } type testHTTPInfrastructureStore struct { - records map[string]*types.HTTPIdempotencyRecord - events []types.BusinessEvent - outbox []types.KafkaOutboxRecord + records map[string]*types.HTTPIdempotencyRecord + events []types.BusinessEvent + deadLetters []types.KafkaDeadLetter } type testTenantEntryStore struct { @@ -162,7 +162,10 @@ func (s *testHTTPInfrastructureStore) MarkKafkaConsumeFailed(_ types.TenantID, e return &types.KafkaConsumeRecord{EventID: eventID, ConsumerGroup: group, Status: types.KafkaConsumeFailed, RetryCount: 1, LastReason: reason, LastError: message}, nil } -func (s *testHTTPInfrastructureStore) MarkKafkaConsumeDead(_ types.TenantID, _ *types.KafkaDeadLetter, _ time.Time) error { +func (s *testHTTPInfrastructureStore) MarkKafkaConsumeDead(_ types.TenantID, dead *types.KafkaDeadLetter, _ time.Time) error { + if dead != nil { + s.deadLetters = append(s.deadLetters, *dead) + } return nil } @@ -182,23 +185,6 @@ func (s *testHTTPInfrastructureStore) MarkKafkaDeadLetterReplay(_ types.TenantID return &types.KafkaDeadLetter{DeadID: deadID, Status: types.KafkaDeadStatusPendingReplay}, nil } -func (s *testHTTPInfrastructureStore) CreateKafkaOutbox(_ types.TenantID, record *types.KafkaOutboxRecord) error { - s.outbox = append(s.outbox, *record) - return nil -} - -func (s *testHTTPInfrastructureStore) ListPendingKafkaOutbox(_ types.TenantID, _ int, _ time.Time) ([]types.KafkaOutboxRecord, error) { - return append([]types.KafkaOutboxRecord(nil), s.outbox...), nil -} - -func (s *testHTTPInfrastructureStore) MarkKafkaOutboxSent(_ types.TenantID, _ string, _ time.Time) error { - return nil -} - -func (s *testHTTPInfrastructureStore) MarkKafkaOutboxFailed(_ types.TenantID, _, _, _ string, _, _ time.Time) error { - return nil -} - type testRelationsStore struct { context *types.RelationshipContext created *types.FriendRequest diff --git a/server/kafka_admin_commands.go b/server/kafka_admin_commands.go index ab086c7..db069d1 100644 --- a/server/kafka_admin_commands.go +++ b/server/kafka_admin_commands.go @@ -38,6 +38,8 @@ const ( adminCommandKafkaDeadResolve = "kafka.dead_letter.resolve" ) +var publishKafkaDeadLetter = publishKafkaEnvelope + type adminCommandPayload struct { CommandID string `json:"command_id"` ActorUserID string `json:"actor_user_id"` @@ -482,6 +484,13 @@ func kafkaTenantID(tenantCode string) (types.TenantID, error) { return tenant.ID, nil } +func kafkaTenantIDFromEnvelope(envelope kafkaEnvelope) (types.TenantID, error) { + if !envelope.TenantID.IsZero() { + return envelope.TenantID, nil + } + return kafkaTenantID(envelope.TenantCode) +} + func markKafkaDeadLetter(ctx context.Context, tenantID types.TenantID, envelope kafkaEnvelope, consume *types.KafkaConsumeRecord, reason, message string) error { encoded, err := json.Marshal(envelope) if err != nil { @@ -512,26 +521,15 @@ func markKafkaDeadLetter(ctx context.Context, tenantID types.TenantID, envelope Status: types.KafkaDeadStatusDead, } now := time.Now().UTC().Round(time.Millisecond) - if err = store.HTTPInfrastructure.MarkKafkaConsumeDead(tenantID, dead, now); err != nil { - return err - } - if auditErr := enqueueKafkaAuditEvent(ctx, tenantID, "system", types.ZeroUid, "kafka.dead_letter.created", "kafka_event", envelope.EventID, "dead", reason, map[string]any{ - "dead_id": deadID, - "event_type": envelope.EventType, - "biz_type": envelope.BizType, - "biz_id": envelope.BizID, - "consumer_group": kafkaBus.config.ConsumerGroup, - }); auditErr != nil { - logs.Warn.Printf("kafka dead letter audit enqueue failed: event=%s dead=%s err=%v", envelope.EventID, deadID, auditErr) - } deadPayload, err := json.Marshal(dead) if err != nil { return err } - return publishKafkaEnvelope(ctx, "dead_letter", envelope.EventID, kafkaEnvelope{ + return publishKafkaDeadLetter(ctx, "dead_letter", envelope.EventID, kafkaEnvelope{ EventID: deadID, TraceID: envelope.TraceID, TenantCode: envelope.TenantCode, + TenantID: tenantID, SourceService: "im-server", TargetService: "hotgo-admin", EventType: "kafka.dead_letter.created", diff --git a/server/kafka_dead_letters.go b/server/kafka_dead_letters.go index 3b8f7d7..dc401a3 100644 --- a/server/kafka_dead_letters.go +++ b/server/kafka_dead_letters.go @@ -6,10 +6,57 @@ import ( "strings" "time" + "github.com/segmentio/kafka-go" + "github.com/tinode/chat/server/logs" "github.com/tinode/chat/server/store" "github.com/tinode/chat/server/store/types" ) +func startKafkaDeadLetterConsumer() { + startKafkaConsumer("dead_letter", handleKafkaDeadLetter) +} + +func handleKafkaDeadLetter(ctx context.Context, envelope kafkaEnvelope, _ kafka.Message) error { + var dead types.KafkaDeadLetter + if err := json.Unmarshal(envelope.Payload, &dead); err != nil { + return err + } + if dead.DeadID == "" { + dead.DeadID = envelope.EventID + } + tenantID, err := tenantIDFromDeadLetter(envelope, &dead) + if err != nil { + return err + } + now := time.Now().UTC().Round(time.Millisecond) + if err = store.HTTPInfrastructure.MarkKafkaConsumeDead(tenantID, &dead, now); err != nil { + return err + } + if auditErr := enqueueKafkaAuditEvent(ctx, tenantID, "system", types.ZeroUid, "kafka.dead_letter.created", "kafka_event", dead.EventID, "dead", dead.LastReason, map[string]any{ + "dead_id": dead.DeadID, + "event_type": dead.EventType, + "biz_type": dead.BizType, + "biz_id": dead.BizID, + "consumer_group": dead.ConsumerGroup, + }); auditErr != nil { + logs.Warn.Printf("kafka dead letter audit enqueue failed: event=%s dead=%s err=%v", dead.EventID, dead.DeadID, auditErr) + } + return nil +} + +func tenantIDFromDeadLetter(envelope kafkaEnvelope, dead *types.KafkaDeadLetter) (types.TenantID, error) { + if !envelope.TenantID.IsZero() { + return envelope.TenantID, nil + } + if dead != nil && len(dead.Envelope) > 0 { + var original kafkaEnvelope + if err := json.Unmarshal(dead.Envelope, &original); err == nil && !original.TenantID.IsZero() { + return original.TenantID, nil + } + } + return kafkaTenantID(envelope.TenantCode) +} + func replayKafkaDeadLetter(ctx context.Context, tenantID types.TenantID, deadID string, actor types.Uid, note string) (*types.KafkaDeadLetter, error) { now := time.Now().UTC().Round(time.Millisecond) dead, err := store.HTTPInfrastructure.MarkKafkaDeadLetterReplay(tenantID, deadID, actor.UserId(), note, now) diff --git a/server/kafka_file_lifecycle.go b/server/kafka_file_lifecycle.go index 38a2370..8bceb36 100644 --- a/server/kafka_file_lifecycle.go +++ b/server/kafka_file_lifecycle.go @@ -104,15 +104,7 @@ func enqueueKafkaGroupFileLifecycle(ctx context.Context, tenantID types.TenantID return enqueueKafkaFileLifecycle(ctx, tenantID, eventType, firstNonEmpty(file.GroupFileID, file.FileID), payload) } -func enqueueKafkaFileLifecycle(_ context.Context, tenantID types.TenantID, eventType, bizID string, payload kafkaFileLifecyclePayload) error { - tenantCode := "" - if store.Tenants != nil { - if tenant, err := store.Tenants.GetByID(tenantID); err == nil && tenant != nil { - tenantCode = tenant.Code - } else { - logs.Warn.Printf("kafka file lifecycle: failed to resolve tenant code for tenant %d: %v", tenantID, err) - } - } +func enqueueKafkaFileLifecycle(ctx context.Context, tenantID types.TenantID, eventType, bizID string, payload kafkaFileLifecyclePayload) error { if payload.OccurredAt.IsZero() { payload.OccurredAt = time.Now().UTC().Round(time.Millisecond) } @@ -121,9 +113,9 @@ func enqueueKafkaFileLifecycle(_ context.Context, tenantID types.TenantID, event return err } eventID := stableIMEventID("file-lifecycle", tenantID, eventType, bizID, payload.SeqID, strings.Join(payload.AttachmentURLs, ",")) - err = enqueueKafkaOutbox(tenantID, "file_lifecycle", bizID, kafkaEnvelope{ + err = publishKafkaEnvelope(ctx, "file_lifecycle", bizID, kafkaEnvelope{ EventID: eventID, - TenantCode: tenantCode, + TenantID: tenantID, SourceService: "im-server", TargetService: "im-server", EventType: eventType, @@ -149,7 +141,7 @@ func enqueueKafkaFileLifecycle(_ context.Context, tenantID types.TenantID, event } func handleKafkaFileLifecycle(ctx context.Context, envelope kafkaEnvelope, _ kafka.Message) error { - tenantID, err := kafkaTenantID(envelope.TenantCode) + tenantID, err := kafkaTenantIDFromEnvelope(envelope) if err != nil { return err } diff --git a/server/kafka_message_persist.go b/server/kafka_message_persist.go index c0fe38a..70c5451 100644 --- a/server/kafka_message_persist.go +++ b/server/kafka_message_persist.go @@ -20,6 +20,8 @@ type kafkaMessagePersistPayload struct { MessageKind string `json:"message_kind"` } +var publishKafkaMessagePersist = publishKafkaEnvelope + func startKafkaMessagePersistConsumer() { startKafkaConsumer("message_persist", handleKafkaMessagePersist) } @@ -28,28 +30,27 @@ func saveMessageMaybeKafka(ctx context.Context, tenantID types.TenantID, msg *ty if !kafkaEnabled() { return store.Messages.Save(tenantID, msg, attachmentURLs, readBySender) } - saveResult, err := store.PrepareMessagePersist(tenantID, msg, attachmentURLs, readBySender) - if err != nil || saveResult == nil || saveResult.Duplicate { - return saveResult, err + if msg == nil || msg.TenantID != tenantID || msg.Topic == "" || msg.SeqId <= 0 { + return nil, types.ErrMalformed } - if readBySender && !types.ParseUid(msg.From).IsZero() { - saveResult.MarkedReadBySender = true + clientMsgID, err := store.NormalizeClientMsgID(msg.Head) + if err != nil { + return nil, err } + msg.ClientMsgID = clientMsgID + msg.InitTimes() if err = enqueueKafkaMessagePersist(ctx, tenantID, msg, attachmentURLs, readBySender); err != nil { return nil, err } - return saveResult, nil -} - -func enqueueKafkaMessagePersist(_ context.Context, tenantID types.TenantID, msg *types.Message, attachmentURLs []string, readBySender bool) error { - tenantCode := "" - if store.Tenants != nil { - if tenant, err := store.Tenants.GetByID(tenantID); err == nil && tenant != nil { - tenantCode = tenant.Code - } else { - logs.Warn.Printf("kafka message persist: failed to resolve tenant code for tenant %d: %v", tenantID, err) + if types.GetTopicCat(msg.Topic) == types.TopicCatGrp { + if cacheErr := store.CachePendingHotMessage(tenantID, msg); cacheErr != nil { + logs.Warn.Printf("kafka message persist: failed to cache pending hot message after publish topic=%s seq=%d: %v", msg.Topic, msg.SeqId, cacheErr) } } + return &types.MessageSaveResult{SeqId: msg.SeqId, MarkedReadBySender: readBySender && !types.ParseUid(msg.From).IsZero()}, nil +} + +func enqueueKafkaMessagePersist(ctx context.Context, tenantID types.TenantID, msg *types.Message, attachmentURLs []string, readBySender bool) error { kind := "p2p" if types.GetTopicCat(msg.Topic) == types.TopicCatGrp { kind = "group" @@ -66,7 +67,7 @@ func enqueueKafkaMessagePersist(_ context.Context, tenantID types.TenantID, msg eventID := fmt.Sprintf("msg.persist.%d.%s.%d", tenantID, msg.Topic, msg.SeqId) envelope := kafkaEnvelope{ EventID: eventID, - TenantCode: tenantCode, + TenantID: tenantID, SourceService: "im-server", TargetService: "im-server", EventType: "message.persist." + kind, @@ -78,13 +79,17 @@ func enqueueKafkaMessagePersist(_ context.Context, tenantID types.TenantID, msg Actor: &kafkaActor{Type: "user", ID: types.ParseUid(msg.From).UserId()}, Payload: payload, } - return enqueueKafkaOutbox(tenantID, "message_persist", msg.Topic, envelope) + return publishKafkaMessagePersist(ctx, "message_persist", msg.Topic, envelope) } func handleKafkaMessagePersist(ctx context.Context, envelope kafkaEnvelope, _ kafka.Message) error { - tenantID, err := kafkaTenantID(envelope.TenantCode) - if err != nil { - return err + tenantID := envelope.TenantID + var err error + if tenantID.IsZero() { + tenantID, err = kafkaTenantID(envelope.TenantCode) + if err != nil { + return err + } } consume, claimed, err := beginKafkaConsumeForTopic(tenantID, envelope, "message_persist") if err != nil || !claimed { diff --git a/server/kafka_outbox.go b/server/kafka_outbox.go deleted file mode 100644 index 993e42c..0000000 --- a/server/kafka_outbox.go +++ /dev/null @@ -1,126 +0,0 @@ -package main - -import ( - "context" - "encoding/json" - "time" - - "github.com/segmentio/kafka-go" - "github.com/tinode/chat/server/logs" - "github.com/tinode/chat/server/store" - "github.com/tinode/chat/server/store/types" -) - -const ( - kafkaOutboxPollInterval = 2 * time.Second - kafkaOutboxTenantLimit = 1000 - kafkaOutboxBatchLimit = 100 -) - -func startKafkaOutboxPoller() { - if !kafkaEnabled() { - return - } - kafkaBus.wg.Add(1) - go func() { - defer kafkaBus.wg.Done() - ticker := time.NewTicker(kafkaOutboxPollInterval) - defer ticker.Stop() - for { - select { - case <-kafkaBus.ctx.Done(): - return - default: - drainKafkaOutbox(kafkaBus.ctx) - } - select { - case <-kafkaBus.ctx.Done(): - return - case <-ticker.C: - } - } - }() -} - -func enqueueKafkaOutbox(tenantID types.TenantID, topicName, key string, envelope kafkaEnvelope) error { - if envelope.OccurredAt.IsZero() { - envelope.OccurredAt = time.Now().UTC() - } - raw, err := json.Marshal(envelope) - if err != nil { - return err - } - return store.HTTPInfrastructure.CreateKafkaOutbox(tenantID, &types.KafkaOutboxRecord{ - EventID: envelope.EventID, - Topic: topicName, - EventType: envelope.EventType, - BizType: envelope.BizType, - BizID: envelope.BizID, - Key: key, - Envelope: raw, - Status: types.KafkaOutboxPending, - NextRetryAt: nil, - }) -} - -func drainKafkaOutbox(ctx context.Context) { - if store.Tenants == nil || store.HTTPInfrastructure == nil { - return - } - tenants, err := store.Tenants.ListActive(kafkaOutboxTenantLimit) - if err != nil { - logs.Warn.Printf("kafka outbox: list active tenants: %v", err) - return - } - now := time.Now().UTC().Round(time.Millisecond) - for _, tenant := range tenants { - if tenant.ID.IsZero() { - continue - } - records, err := store.HTTPInfrastructure.ListPendingKafkaOutbox(tenant.ID, kafkaOutboxBatchLimit, now) - if err != nil { - logs.Warn.Printf("kafka outbox: list tenant=%d: %v", tenant.ID, err) - continue - } - for _, record := range records { - if err = publishKafkaOutboxRecord(ctx, record); err != nil { - nextRetryAt := now.Add(kafkaOutboxRetryDelay(record.Attempts + 1)) - if markErr := store.HTTPInfrastructure.MarkKafkaOutboxFailed(tenant.ID, record.EventID, - "KAFKA_PUBLISH_FAILED", err.Error(), nextRetryAt, now); markErr != nil { - logs.Warn.Printf("kafka outbox: mark failed event=%s tenant=%d: %v", record.EventID, tenant.ID, markErr) - } - continue - } - if err = store.HTTPInfrastructure.MarkKafkaOutboxSent(tenant.ID, record.EventID, time.Now().UTC().Round(time.Millisecond)); err != nil { - logs.Warn.Printf("kafka outbox: mark sent event=%s tenant=%d: %v", record.EventID, tenant.ID, err) - } - } - } -} - -func publishKafkaOutboxRecord(ctx context.Context, record types.KafkaOutboxRecord) error { - topic := kafkaTopic(record.Topic) - if topic == "" { - topic = record.Topic - } - return kafkaBus.writer.WriteMessages(ctx, kafka.Message{ - Topic: topic, - Key: []byte(record.Key), - Value: record.Envelope, - Time: time.Now().UTC(), - }) -} - -func kafkaOutboxRetryDelay(attempt int) time.Duration { - if attempt <= 0 { - attempt = 1 - } - if kafkaBus == nil || len(kafkaBus.config.RetryBackoffMS) == 0 { - return time.Duration(attempt) * 10 * time.Second - } - index := attempt - 1 - if index >= len(kafkaBus.config.RetryBackoffMS) { - index = len(kafkaBus.config.RetryBackoffMS) - 1 - } - return time.Duration(kafkaBus.config.RetryBackoffMS[index]) * time.Millisecond -} diff --git a/server/kafka_runtime.go b/server/kafka_runtime.go index c306b21..5ab83f4 100644 --- a/server/kafka_runtime.go +++ b/server/kafka_runtime.go @@ -13,6 +13,7 @@ import ( "github.com/segmentio/kafka-go/sasl" "github.com/segmentio/kafka-go/sasl/scram" "github.com/tinode/chat/server/logs" + "github.com/tinode/chat/server/store/types" ) const ( @@ -55,6 +56,7 @@ type kafkaEnvelope struct { EventID string `json:"event_id"` TraceID string `json:"trace_id,omitempty"` TenantCode string `json:"tenant_code,omitempty"` + TenantID types.TenantID `json:"tenant_id,omitempty"` SourceService string `json:"source_service,omitempty"` TargetService string `json:"target_service,omitempty"` EventType string `json:"event_type"` @@ -132,6 +134,7 @@ func initKafkaRuntime(raw json.RawMessage) error { AllowAutoTopicCreation: cfg.AllowAutoTopic, WriteTimeout: time.Duration(cfg.WriteTimeoutMS) * time.Millisecond, ReadTimeout: time.Duration(cfg.ReadTimeoutMS) * time.Millisecond, + MaxAttempts: cfg.MaxRetryCount, } if err := rt.ping(context.Background()); err != nil { @@ -164,7 +167,7 @@ func normalizeKafkaConfig(cfg *kafkaConfig) { cfg.RequiredAcks = int(kafka.RequireOne) } if cfg.MaxRetryCount <= 0 { - cfg.MaxRetryCount = 5 + cfg.MaxRetryCount = 3 } if len(cfg.RetryBackoffMS) == 0 { cfg.RetryBackoffMS = []int{10000, 30000, 60000, 300000, 900000} diff --git a/server/kafka_workbench_audit.go b/server/kafka_workbench_audit.go index 4ab18b2..1f8b845 100644 --- a/server/kafka_workbench_audit.go +++ b/server/kafka_workbench_audit.go @@ -6,7 +6,6 @@ import ( "time" "github.com/segmentio/kafka-go" - "github.com/tinode/chat/server/logs" "github.com/tinode/chat/server/store" "github.com/tinode/chat/server/store/types" ) @@ -87,15 +86,7 @@ func enqueueKafkaAuditEvent(ctx context.Context, tenantID types.TenantID, actorT return enqueueKafkaTypedEvent(ctx, tenantID, "audit_event", "audit."+action, "audit", resource+":"+resourceID, actor, payload) } -func enqueueKafkaTypedEvent(_ context.Context, tenantID types.TenantID, topicName, eventType, bizType, bizID string, actor types.Uid, payload any) error { - tenantCode := "" - if store.Tenants != nil { - if tenant, err := store.Tenants.GetByID(tenantID); err == nil && tenant != nil { - tenantCode = tenant.Code - } else { - logs.Warn.Printf("kafka event: failed to resolve tenant code for tenant %d: %v", tenantID, err) - } - } +func enqueueKafkaTypedEvent(ctx context.Context, tenantID types.TenantID, topicName, eventType, bizType, bizID string, actor types.Uid, payload any) error { encoded, err := json.Marshal(payload) if err != nil { return err @@ -103,7 +94,7 @@ func enqueueKafkaTypedEvent(_ context.Context, tenantID types.TenantID, topicNam eventID := stableIMEventID(topicName, tenantID, eventType, bizID, encoded) envelope := kafkaEnvelope{ EventID: eventID, - TenantCode: tenantCode, + TenantID: tenantID, SourceService: "im-server", TargetService: "im-server", EventType: eventType, @@ -115,11 +106,11 @@ func enqueueKafkaTypedEvent(_ context.Context, tenantID types.TenantID, topicNam if !actor.IsZero() { envelope.Actor = &kafkaActor{Type: "user", ID: actor.UserId()} } - return enqueueKafkaOutbox(tenantID, topicName, bizID, envelope) + return publishKafkaEnvelope(ctx, topicName, bizID, envelope) } func handleKafkaWorkbenchEvent(ctx context.Context, envelope kafkaEnvelope, _ kafka.Message) error { - tenantID, err := kafkaTenantID(envelope.TenantCode) + tenantID, err := kafkaTenantIDFromEnvelope(envelope) if err != nil { return err } diff --git a/server/main.go b/server/main.go index cb89653..913aa5b 100644 --- a/server/main.go +++ b/server/main.go @@ -698,7 +698,7 @@ func main() { startKafkaMessagePersistConsumer() startKafkaFileLifecycleConsumer() startKafkaWorkbenchEventConsumer() - startKafkaOutboxPoller() + startKafkaDeadLetterConsumer() stopExternalBusinessEvents := startIMExternalBusinessEventPoller() defer stopExternalBusinessEvents() diff --git a/server/meetingim.conf b/server/meetingim.conf index 02ae0fb..4dfa15b 100644 --- a/server/meetingim.conf +++ b/server/meetingim.conf @@ -95,7 +95,7 @@ "dial_timeout_ms": 2000, "read_timeout_ms": 2000, "write_timeout_ms": 2000, - "pending_ttl_seconds": 120, + "pending_ttl_seconds": 300, "done_ttl_seconds": 86400, "failed_ttl_seconds": 600, "hot_message_ttl_seconds": 86400, @@ -121,7 +121,7 @@ "sasl_mechanism": "SCRAM-SHA-512", "sasl_username": "kafka-external", "sasl_password": "gr58tNnlZOVGW3AmMbL2", - "max_retry_count": 5, + "max_retry_count": 3, "retry_backoff_ms": [10000, 30000, 60000, 300000, 900000], "dead_letter_retention_seconds": 7776000, "topics": { diff --git a/server/store/group_message_runtime.go b/server/store/group_message_runtime.go index ad1a4e3..b8d38e1 100644 --- a/server/store/group_message_runtime.go +++ b/server/store/group_message_runtime.go @@ -51,6 +51,7 @@ type groupMessageRuntime interface { MarkDone(tenantID types.TenantID, topic string, from types.Uid, clientMsgID string, seq int64, msgID string) error MarkFailed(tenantID types.TenantID, topic string, from types.Uid, clientMsgID string) error CacheHotMessage(tenantID types.TenantID, msg *types.Message) error + CachePendingHotMessage(tenantID types.TenantID, msg *types.Message) error GetHotMessages(tenantID types.TenantID, topic string, opts *types.QueryOpt, maxLimit int) ([]types.Message, bool, error) InvalidateHotMessages(tenantID types.TenantID, topic string, ranges []types.Range) error MarkUserOnline(tenantID types.TenantID, uid types.Uid, sid, nodeID, device, userAgent string) error @@ -187,6 +188,27 @@ func DeleteCachedObject(tenantID types.TenantID, namespace, id string) error { return groupMsgRuntime.DeleteCachedObject(tenantID, namespace, id) } +func CacheHotMessage(tenantID types.TenantID, msg *types.Message) error { + if groupMsgRuntime == nil || !tenantID.IsValid() || msg == nil || msg.Topic == "" || msg.SeqId <= 0 { + return nil + } + return groupMsgRuntime.CacheHotMessage(tenantID, msg) +} + +func CachePendingHotMessage(tenantID types.TenantID, msg *types.Message) error { + if groupMsgRuntime == nil || !tenantID.IsValid() || msg == nil || msg.Topic == "" || msg.SeqId <= 0 { + return nil + } + return groupMsgRuntime.CachePendingHotMessage(tenantID, msg) +} + +func GetHotMessages(tenantID types.TenantID, topic string, opts *types.QueryOpt, maxLimit int) ([]types.Message, bool, error) { + if groupMsgRuntime == nil || !tenantID.IsValid() || topic == "" { + return nil, false, nil + } + return groupMsgRuntime.GetHotMessages(tenantID, topic, opts, maxLimit) +} + func InvalidateHotMessages(tenantID types.TenantID, topic string, ranges []types.Range) error { if groupMsgRuntime == nil || !tenantID.IsValid() || topic == "" { return nil @@ -239,7 +261,7 @@ func initGroupMessageRuntime(raw json.RawMessage) error { cfg.KeyPrefix = "im" } if cfg.PendingTTL <= 0 { - cfg.PendingTTL = 120 + cfg.PendingTTL = 300 } if cfg.DoneTTL <= 0 { cfg.DoneTTL = 86400 @@ -380,15 +402,25 @@ func (r *redisGroupMessageRuntime) MarkFailed(tenantID types.TenantID, topic str } func (r *redisGroupMessageRuntime) CacheHotMessage(tenantID types.TenantID, msg *types.Message) error { + return r.cacheHotMessage(tenantID, msg, time.Duration(r.cfg.HotMsgTTL)*time.Second) +} + +func (r *redisGroupMessageRuntime) CachePendingHotMessage(tenantID types.TenantID, msg *types.Message) error { + return r.cacheHotMessage(tenantID, msg, time.Duration(r.cfg.PendingTTL)*time.Second) +} + +func (r *redisGroupMessageRuntime) cacheHotMessage(tenantID types.TenantID, msg *types.Message, ttl time.Duration) error { if msg == nil || msg.Topic == "" || msg.SeqId <= 0 { return types.ErrMalformed } + if ttl <= 0 { + return types.ErrMalformed + } payload, err := json.Marshal(msg) if err != nil { return err } ctx := context.Background() - ttl := time.Duration(r.cfg.HotMsgTTL) * time.Second seq := strconv.FormatInt(msg.SeqId, 10) pipe := r.client.Pipeline() pipe.Set(ctx, r.hotMsgKey(tenantID, msg.Topic, msg.SeqId), payload, ttl) diff --git a/server/store/infrastructure.go b/server/store/infrastructure.go index e77b398..52a9a18 100644 --- a/server/store/infrastructure.go +++ b/server/store/infrastructure.go @@ -22,10 +22,6 @@ type HTTPInfrastructurePersistenceInterface interface { GetKafkaDeadLetter(types.TenantID, string) (*types.KafkaDeadLetter, error) MarkKafkaDeadLetter(types.TenantID, string, string, string, string, time.Time) error MarkKafkaDeadLetterReplay(types.TenantID, string, string, string, time.Time) (*types.KafkaDeadLetter, error) - CreateKafkaOutbox(types.TenantID, *types.KafkaOutboxRecord) error - ListPendingKafkaOutbox(types.TenantID, int, time.Time) ([]types.KafkaOutboxRecord, error) - MarkKafkaOutboxSent(types.TenantID, string, time.Time) error - MarkKafkaOutboxFailed(types.TenantID, string, string, string, time.Time, time.Time) error } type httpInfrastructureMapper struct{} @@ -155,35 +151,3 @@ func (httpInfrastructureMapper) MarkKafkaDeadLetterReplay(tenantID types.TenantI } return infrastructure.TenantKafkaDeadLetterMarkReplay(tenantID, deadID, resolvedBy, note, now) } - -func (httpInfrastructureMapper) CreateKafkaOutbox(tenantID types.TenantID, record *types.KafkaOutboxRecord) error { - infrastructure, err := tenantHTTPInfrastructureAdapter(tenantID) - if err != nil { - return err - } - return infrastructure.TenantKafkaOutboxCreate(tenantID, record) -} - -func (httpInfrastructureMapper) ListPendingKafkaOutbox(tenantID types.TenantID, limit int, now time.Time) ([]types.KafkaOutboxRecord, error) { - infrastructure, err := tenantHTTPInfrastructureAdapter(tenantID) - if err != nil { - return nil, err - } - return infrastructure.TenantKafkaOutboxListPending(tenantID, limit, now) -} - -func (httpInfrastructureMapper) MarkKafkaOutboxSent(tenantID types.TenantID, eventID string, now time.Time) error { - infrastructure, err := tenantHTTPInfrastructureAdapter(tenantID) - if err != nil { - return err - } - return infrastructure.TenantKafkaOutboxMarkSent(tenantID, eventID, now) -} - -func (httpInfrastructureMapper) MarkKafkaOutboxFailed(tenantID types.TenantID, eventID, reason, message string, nextRetryAt, now time.Time) error { - infrastructure, err := tenantHTTPInfrastructureAdapter(tenantID) - if err != nil { - return err - } - return infrastructure.TenantKafkaOutboxMarkFailed(tenantID, eventID, reason, message, nextRetryAt, now) -} diff --git a/server/store/store.go b/server/store/store.go index 1a67aee..cf49574 100644 --- a/server/store/store.go +++ b/server/store/store.go @@ -3443,6 +3443,10 @@ func PersistPreparedMessage(tenantID types.TenantID, msg *types.Message, attachm return Messages.PersistPrepared(tenantID, msg, attachmentURLs, readBySender) } +func NormalizeClientMsgID(head types.KVMap) (string, error) { + return normalizeClientMsgID(head) +} + func retryTenantMessageSaveAfterSeqConflict(tenantAdp adapter.TenantBusinessAdapter, tenantID types.TenantID, msg *types.Message) error { latest, err := tenantAdp.TenantMessageGetAll(tenantID, msg.Topic, types.ZeroUid, &types.QueryOpt{Before: 1 << 62, Limit: 1}) if err != nil { diff --git a/server/store/types/infrastructure.go b/server/store/types/infrastructure.go index 18dfe02..261a9a2 100644 --- a/server/store/types/infrastructure.go +++ b/server/store/types/infrastructure.go @@ -23,10 +23,6 @@ const ( KafkaDeadStatusReplayed = "replayed" KafkaDeadStatusIgnored = "ignored" KafkaDeadStatusResolved = "resolved" - - KafkaOutboxPending = "pending" - KafkaOutboxSent = "sent" - KafkaOutboxFailed = "failed" ) // HTTPIdempotencyRecord stores the first response for one authenticated write. @@ -115,26 +111,6 @@ type KafkaDeadLetterPage struct { NextCursor string } -// KafkaOutboxRecord stores a Kafka event waiting for reliable publication. -type KafkaOutboxRecord struct { - ID int64 - EventID string - Topic string - EventType string - BizType string - BizID string - Key string - Envelope []byte - Status string - Attempts int - NextRetryAt *time.Time - SentAt *time.Time - LastReason string - LastError string - CreatedAt time.Time - UpdatedAt time.Time -} - // StableEventID derives an idempotent event identifier from domain identity. func StableEventID(parts ...any) string { hash := sha256.New() diff --git a/server/tinode.conf b/server/tinode.conf index 02ae0fb..4dfa15b 100644 --- a/server/tinode.conf +++ b/server/tinode.conf @@ -95,7 +95,7 @@ "dial_timeout_ms": 2000, "read_timeout_ms": 2000, "write_timeout_ms": 2000, - "pending_ttl_seconds": 120, + "pending_ttl_seconds": 300, "done_ttl_seconds": 86400, "failed_ttl_seconds": 600, "hot_message_ttl_seconds": 86400, @@ -121,7 +121,7 @@ "sasl_mechanism": "SCRAM-SHA-512", "sasl_username": "kafka-external", "sasl_password": "gr58tNnlZOVGW3AmMbL2", - "max_retry_count": 5, + "max_retry_count": 3, "retry_backoff_ms": [10000, 30000, 60000, 300000, 900000], "dead_letter_retention_seconds": 7776000, "topics": { diff --git a/server/topic.go b/server/topic.go index 5ffda32..dbcd0a1 100644 --- a/server/topic.go +++ b/server/topic.go @@ -2984,11 +2984,20 @@ func (t *Topic) replyGetData(sess *Session, asUid types.Uid, asChan bool, req *M count := 0 if userData := t.perUser[asUid]; (userData.modeGiven & userData.modeWant).IsReader() { // Read messages from DB - messages, err := store.Messages.GetAll(t.tenantID, t.name, asUid, msgOpts2storeOpts(req)) + opts := msgOpts2storeOpts(req) + messages, err := store.Messages.GetAll(t.tenantID, t.name, asUid, opts) if err != nil { sess.queueOut(ErrUnknownReply(msg, now)) return err } + if len(messages) == 0 && shouldTryHotMessageFallback(t.cat, t.lastID, opts) { + hotMessages, ok, cacheErr := store.GetHotMessages(t.tenantID, t.name, opts, hotMessageFallbackLimit(opts)) + if cacheErr != nil { + logs.Warn.Printf("get.data hot message fallback failed topic=%s user=%s: %v", t.name, asUid.UserId(), cacheErr) + } else if ok { + messages = filterHotMessagesForUser(hotMessages, asUid) + } + } // Push the list of messages to the client as {data}. if messages != nil { @@ -3032,6 +3041,56 @@ func (t *Topic) replyGetData(sess *Session, asUid types.Uid, asChan bool, req *M return nil } +func shouldTryHotMessageFallback(cat types.TopicCat, lastID int64, opts *types.QueryOpt) bool { + if cat != types.TopicCatGrp || opts == nil || opts.IfModifiedSince != nil || + !opts.User.IsZero() || opts.Topic != "" || opts.Since <= 0 || opts.Before > 0 || len(opts.IdRanges) > 0 { + return false + } + limit := hotMessageFallbackLimit(opts) + if limit <= 0 || limit > 10 { + return false + } + floor := lastID - 10 + if floor < 1 { + floor = 1 + } + return lastID <= 0 || opts.Since >= floor +} + +func hotMessageFallbackLimit(opts *types.QueryOpt) int { + if opts != nil && opts.Limit > 0 { + return opts.Limit + } + return 1 +} + +func filterHotMessagesForUser(messages []types.Message, asUid types.Uid) []types.Message { + if len(messages) == 0 { + return messages + } + filtered := messages[:0] + for i := range messages { + if hotMessageVisibleToUser(&messages[i], asUid) { + filtered = append(filtered, messages[i]) + } + } + return filtered +} + +func hotMessageVisibleToUser(msg *types.Message, asUid types.Uid) bool { + if msg == nil || msg.DeletedAt != nil || msg.DelId > 0 || asUid.IsZero() { + return msg != nil && msg.DeletedAt == nil && msg.DelId == 0 + } + uid := asUid.String() + userID := asUid.UserId() + for _, deleted := range msg.DeletedFor { + if deleted.User == uid || deleted.User == userID { + return false + } + } + return true +} + // replyGetTags returns topics' tags - tokens used for discovery. func (t *Topic) replyGetTags(sess *Session, asUid types.Uid, msg *ClientComMessage) error { now := types.TimeNow() diff --git a/server/topic_test.go b/server/topic_test.go index b514492..1f1d211 100644 --- a/server/topic_test.go +++ b/server/topic_test.go @@ -1,6 +1,8 @@ package main import ( + "context" + "encoding/json" "fmt" "net/http" "os" @@ -9,6 +11,7 @@ import ( "time" "github.com/golang/mock/gomock" + "github.com/segmentio/kafka-go" "github.com/tinode/chat/server/auth" "github.com/tinode/chat/server/logs" "github.com/tinode/chat/server/store" @@ -901,6 +904,196 @@ func TestHandleBroadcastDataFiltersInvalidMentions(t *testing.T) { verifyResponseCodes(helper.results[0], []int{http.StatusAccepted}, t) } +func TestSaveMessageMaybeKafkaPublishesWithoutDBPersist(t *testing.T) { + ctrl := gomock.NewController(t) + defer ctrl.Finish() + + oldKafkaBus := kafkaBus + oldMessages := store.Messages + oldPublisher := publishKafkaMessagePersist + defer func() { + kafkaBus = oldKafkaBus + store.Messages = oldMessages + publishKafkaMessagePersist = oldPublisher + }() + + kafkaBus = &kafkaRuntime{} + store.Messages = mock_store.NewMockMessagesPersistenceInterface(ctrl) + var publishedTopic string + var publishedKey string + var publishedEnvelope kafkaEnvelope + publishKafkaMessagePersist = func(_ context.Context, topicName, key string, envelope kafkaEnvelope) error { + publishedTopic = topicName + publishedKey = key + publishedEnvelope = envelope + return nil + } + + msg := &types.Message{ + TenantID: 1, + Topic: "grp-test", + SeqId: 1, + From: types.Uid(1).String(), + Head: types.KVMap{"x-im-client-mid": "client-1"}, + Content: "hello", + } + + got, err := saveMessageMaybeKafka(context.Background(), 1, msg, []string{"file1"}, true) + if err != nil { + t.Fatalf("saveMessageMaybeKafka returned error: %v", err) + } + if got == nil || got.SeqId != 1 || !got.MarkedReadBySender { + t.Fatalf("unexpected save result: %#v", got) + } + if msg.ClientMsgID != "client-1" { + t.Fatalf("client message id not normalized: %q", msg.ClientMsgID) + } + if publishedTopic != "message_persist" { + t.Fatalf("unexpected kafka topic: %q", publishedTopic) + } + if publishedKey != "grp-test" { + t.Fatalf("unexpected kafka key: %q", publishedKey) + } + if publishedEnvelope.EventType != "message.persist.group" { + t.Fatalf("unexpected event type: %q", publishedEnvelope.EventType) + } + if publishedEnvelope.TenantID != 1 || publishedEnvelope.TenantCode != "" { + t.Fatalf("unexpected tenant routing fields: tenant_id=%d tenant_code=%q", publishedEnvelope.TenantID, publishedEnvelope.TenantCode) + } + if publishedEnvelope.IdempotencyKey != "client-1" { + t.Fatalf("unexpected idempotency key: %q", publishedEnvelope.IdempotencyKey) + } +} + +func TestShouldTryHotMessageFallback(t *testing.T) { + if !shouldTryHotMessageFallback(types.TopicCatGrp, 159, &types.QueryOpt{Since: 159, Limit: 1}) { + t.Fatal("expected fallback for latest group seq query") + } + if shouldTryHotMessageFallback(types.TopicCatP2P, 159, &types.QueryOpt{Since: 159, Limit: 1}) { + t.Fatal("did not expect fallback for p2p topic") + } + if shouldTryHotMessageFallback(types.TopicCatGrp, 159, &types.QueryOpt{Since: 1, Limit: 1}) { + t.Fatal("did not expect fallback for old history query") + } + if shouldTryHotMessageFallback(types.TopicCatGrp, 159, &types.QueryOpt{Since: 159, Limit: 50}) { + t.Fatal("did not expect fallback for broad query") + } +} + +func TestFilterHotMessagesForUserSkipsDeleted(t *testing.T) { + uid := types.Uid(1) + now := time.Now() + messages := []types.Message{ + {SeqId: 1}, + {SeqId: 2, DeletedFor: []types.SoftDelete{{User: uid.String()}}}, + {SeqId: 3, DeletedFor: []types.SoftDelete{{User: uid.UserId()}}}, + {SeqId: 4, DeletedAt: &now}, + {SeqId: 5, DelId: 10}, + } + filtered := filterHotMessagesForUser(messages, uid) + if len(filtered) != 1 || filtered[0].SeqId != 1 { + t.Fatalf("unexpected filtered messages: %#v", filtered) + } +} + +func TestMarkKafkaDeadLetterPublishesBeforeDBWrite(t *testing.T) { + oldKafkaBus := kafkaBus + oldPublisher := publishKafkaDeadLetter + oldInfrastructure := store.HTTPInfrastructure + defer func() { + kafkaBus = oldKafkaBus + publishKafkaDeadLetter = oldPublisher + store.HTTPInfrastructure = oldInfrastructure + }() + + kafkaBus = &kafkaRuntime{config: kafkaConfig{ConsumerGroup: "im-server"}} + infrastructure := &testHTTPInfrastructureStore{} + store.HTTPInfrastructure = infrastructure + var publishedEnvelope kafkaEnvelope + publishKafkaDeadLetter = func(_ context.Context, topicName, key string, envelope kafkaEnvelope) error { + if topicName != "dead_letter" || key != "event-1" { + t.Fatalf("unexpected dead letter publish target topic=%q key=%q", topicName, key) + } + publishedEnvelope = envelope + return nil + } + + original := kafkaEnvelope{ + EventID: "event-1", + TenantID: 1, + EventType: "message.persist.group", + BizType: "message", + BizID: "grp-test:1", + } + consume := &types.KafkaConsumeRecord{ + EventID: "event-1", + ConsumerGroup: "im-server", + SourceTopic: "im.message.persist", + RetryCount: 3, + } + if err := markKafkaDeadLetter(context.Background(), 1, original, consume, "DB_FAILED", "boom"); err != nil { + t.Fatalf("markKafkaDeadLetter returned error: %v", err) + } + if len(infrastructure.deadLetters) != 0 { + t.Fatalf("dead letter was written before dead_letter topic consumption: %#v", infrastructure.deadLetters) + } + if publishedEnvelope.EventID == "" || publishedEnvelope.TenantID != 1 { + t.Fatalf("dead letter event was not published correctly: %#v", publishedEnvelope) + } +} + +func TestHandleKafkaDeadLetterWritesDB(t *testing.T) { + oldKafkaBus := kafkaBus + oldInfrastructure := store.HTTPInfrastructure + defer func() { + kafkaBus = oldKafkaBus + store.HTTPInfrastructure = oldInfrastructure + }() + + kafkaBus = nil + infrastructure := &testHTTPInfrastructureStore{} + store.HTTPInfrastructure = infrastructure + original := kafkaEnvelope{ + EventID: "event-1", + TenantID: 1, + EventType: "message.persist.group", + BizType: "message", + BizID: "grp-test:1", + } + encodedOriginal, _ := json.Marshal(original) + dead := types.KafkaDeadLetter{ + DeadID: "dead-1", + EventID: "event-1", + SourceTopic: "im.message.persist", + ConsumerGroup: "im-server", + EventType: "message.persist.group", + BizType: "message", + BizID: "grp-test:1", + Envelope: encodedOriginal, + RetryCount: 3, + LastReason: "DB_FAILED", + LastError: "boom", + Status: types.KafkaDeadStatusDead, + } + payload, _ := json.Marshal(dead) + envelope := kafkaEnvelope{ + EventID: "dead-1", + TenantID: 1, + EventType: "kafka.dead_letter.created", + ParentEventID: "event-1", + Payload: payload, + } + if err := handleKafkaDeadLetter(context.Background(), envelope, kafka.Message{}); err != nil { + t.Fatalf("handleKafkaDeadLetter returned error: %v", err) + } + if len(infrastructure.deadLetters) != 1 { + t.Fatalf("expected one dead letter write, got %d", len(infrastructure.deadLetters)) + } + if infrastructure.deadLetters[0].DeadID != "dead-1" || infrastructure.deadLetters[0].EventID != "event-1" { + t.Fatalf("unexpected dead letter write: %#v", infrastructure.deadLetters[0]) + } +} + func TestHandleBroadcastDataInactiveTopic(t *testing.T) { numUsers := 2 helper := TopicTestHelper{}