From 659373e1421739d63895dc8ffcfb9fc41a2cb5b4 Mon Sep 17 00:00:00 2001 From: Nixevol Date: Wed, 30 Sep 2026 15:06:21 +0800 Subject: [PATCH] =?UTF-8?q?fix:=20=E8=A1=A5=E6=8F=90=E4=BA=A4=E6=A0=A1?= =?UTF-8?q?=E9=AA=8C=E3=80=81=E5=81=9C=E7=94=A8=E6=A3=80=E6=9F=A5=E3=80=81?= =?UTF-8?q?=E5=85=A5=E7=BE=A4=E8=BF=87=E6=BB=A4=E4=B8=8E=E6=8C=89=E5=AE=8C?= =?UTF-8?q?=E6=88=90=E6=97=B6=E5=88=BB=E6=B8=85=E7=90=86?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- docs/DEVIATIONS.md | 9 ++ internal/app/message/app.go | 8 +- internal/app/message/recover.go | 10 +- internal/app/message/submit.go | 27 +++- internal/app/message/submit_test.go | 136 +++++++++++++++++++++ internal/app/message/void_finalize_test.go | 43 +++++++ internal/protocol/types.go | 8 +- internal/protocol/validate.go | 17 +++ 8 files changed, 247 insertions(+), 11 deletions(-) diff --git a/docs/DEVIATIONS.md b/docs/DEVIATIONS.md index 2ab9575..ee3b21f 100644 --- a/docs/DEVIATIONS.md +++ b/docs/DEVIATIONS.md @@ -518,6 +518,15 @@ - 备选方案：`Meta` 改为 `json.RawMessage` 原样输出（需改 protocol，牵动 SDK）。 - 影响：仅推送路径；入库仍是提交时的规范 JSON。 +### 复审修复 C-07 + +1. **提交校验、停用检查、入群时间过滤、保留期按完成时刻** + - 原条款：DEVELOPMENT 6.2 ttl/定时上限；PRD F01 停用后不能再发；F06 发送时刻之后入群的端收不到；F18 记录保留从完成起算。 + - 实际做法：`keep` 且 `ttl_seconds<=0` 回 `bad_request`；`delay_ms` 先与 `max_schedule_seconds*1000` 比较再加法。`Send.Validate` 同步（`protocol.Limits` 增加可选 MaxTTL/MaxSchedule，0 表示不查上限）。写事务内检查发送方 `enabled`，停用回 `unauthorized`。群分发 `joined_at <= send_at`。未做 C-03 的 `completed_at` 列，清理暂用 `MAX(deliveries.updated_at)` 否则 `send_at` 近似完成时刻。不改对话密码锁键语义。 + - 原因：ttl=0/负数、delay 溢出、停用窗口内仍能提交、晚入群仍能收到、按 created_at 清理会误删长定时/长保留消息。 + - 备选方案：等 C-03 迁移后改用 `completed_at`；发送方停用改用 `endpoint_disabled`（与目标停用混用）。 + - 影响：发送方停用错误码为 `unauthorized`；保留期口径在 C-03 合入前对无投递的 scheduled 作废行用 `send_at` 近似。 + ## 身份 I ### I1 2026-09-30 diff --git a/internal/app/message/app.go b/internal/app/message/app.go index ae9d39f..6e55ea9 100644 --- a/internal/app/message/app.go +++ b/internal/app/message/app.go @@ -162,9 +162,11 @@ func (a *App) now() time.Time { func (a *App) protocolLimits() protocol.Limits { return protocol.Limits{ - MaxBodyBytes: a.lim.MaxBodyBytes, - MaxMetaBytes: a.lim.MaxMetaBytes, - MaxFrameBytes: a.lim.MaxFrameBytes, + MaxBodyBytes: a.lim.MaxBodyBytes, + MaxMetaBytes: a.lim.MaxMetaBytes, + MaxFrameBytes: a.lim.MaxFrameBytes, + MaxTTLSeconds: a.lim.MaxTTLSeconds, + MaxScheduleSeconds: a.lim.MaxScheduleSeconds, } } diff --git a/internal/app/message/recover.go b/internal/app/message/recover.go index 5edabc7..2754928 100644 --- a/internal/app/message/recover.go +++ b/internal/app/message/recover.go @@ -86,7 +86,15 @@ WHERE d.state = 'pending' AND d.pushed_conn IS NULL cutoff := nowMs - int64(a.lim.RecordRetentionDays)*24*3600*1000 if _, err := tx.Exec(` DELETE FROM messages WHERE seq IN ( - SELECT seq FROM messages WHERE state = 'completed' AND created_at < ? LIMIT 5000 + SELECT seq FROM ( + SELECT m.seq FROM messages m + WHERE m.state = 'completed' + AND COALESCE( + (SELECT MAX(d.updated_at) FROM deliveries d WHERE d.seq = m.seq), + m.send_at + ) < ? + LIMIT 5000 + ) )`, cutoff); err != nil { return err } diff --git a/internal/app/message/submit.go b/internal/app/message/submit.go index acf8a96..f69fa83 100644 --- a/internal/app/message/submit.go +++ b/internal/app/message/submit.go @@ -46,6 +46,9 @@ func (a *App) Submit(ctx context.Context, senderID string, conn port.ConnInfo, r keep := protocol.EffectiveOfflineKeep(req) ttl := protocol.EffectiveOfflineTTL(req) receipt := protocol.EffectiveReceipt(req) + if keep && ttl <= 0 { + return SubmitResult{}, errCode(protocol.CodeBadRequest, "ttl_seconds must be > 0") + } if keep && a.lim.MaxTTLSeconds > 0 && ttl > a.lim.MaxTTLSeconds { return SubmitResult{}, errCode(protocol.CodeBadRequest, "ttl_seconds exceeds max_ttl_seconds") } @@ -66,6 +69,9 @@ func (a *App) Submit(ctx context.Context, senderID string, conn port.ConnInfo, r } return SubmitResult{}, err } + if sender.Enabled == 0 { + return SubmitResult{}, errCode(protocol.CodeUnauthorized, "sender disabled") + } sendAt, err := a.computeSendAt(req, sender.DefaultDelayMs, nowMs) if err != nil { @@ -155,6 +161,17 @@ func (a *App) Submit(ctx context.Context, senderID string, conn port.ConnInfo, r return e } + snd, se := loadEndpointTx(tx, senderID) + if se != nil { + if errors.Is(se, sql.ErrNoRows) { + return errCode(protocol.CodeInvalidTarget, "sender not found") + } + return se + } + if snd.Enabled == 0 { + return errCode(protocol.CodeUnauthorized, "sender disabled") + } + // 写事务内再确认目标与授权（防并发停用/退群）。 switch req.To.Kind { case protocol.TargetEndpoint: @@ -197,10 +214,6 @@ func (a *App) Submit(ctx context.Context, senderID string, conn port.ConnInfo, r } } // 发送方设了对话密码且发给别人的单聊：给对方写回复授权。 - snd, se := loadEndpointTx(tx, senderID) - if se != nil { - return se - } if senderID != req.To.ID && snd.TalkHash != nil && *snd.TalkHash != "" { if ge := upsertGrantTx(tx, req.To.ID, senderID, snd.TalkVersion, GrantKindReply, nowMs); ge != nil { return ge @@ -295,6 +308,12 @@ func (a *App) computeSendAt(req *protocol.Send, defaultDelayMs, nowMs int64) (in if *req.DelayMs < 0 { return 0, errCode(protocol.CodeBadRequest, "delay_ms negative") } + if a.lim.MaxScheduleSeconds > 0 { + maxDelay := a.lim.MaxScheduleSeconds * 1000 + if *req.DelayMs > maxDelay { + return 0, errCode(protocol.CodeBadRequest, "send time exceeds max_schedule_seconds") + } + } sendAt = nowMs + *req.DelayMs default: if defaultDelayMs < 0 { diff --git a/internal/app/message/submit_test.go b/internal/app/message/submit_test.go index 208c346..38c3bc4 100644 --- a/internal/app/message/submit_test.go +++ b/internal/app/message/submit_test.go @@ -4,6 +4,7 @@ import ( "context" "database/sql" "errors" + "math" "path/filepath" "testing" "time" @@ -363,6 +364,141 @@ SELECT kind FROM talk_grants WHERE sender_id=? AND target_id=?`, "bob", "alice") } }) + t.Run("ttl_zero_rejected", func(t *testing.T) { + t.Parallel() + lim := defaultTestLimits() + app, db := openTestApp(t, lim) + insertEndpoint(t, db, "alice", "", 1, 0) + insertEndpoint(t, db, "bob", "", 1, 0) + ttl := int64(0) + req := baseSend("ttl0", "bob") + req.Offline = &protocol.OfflineOpts{Keep: true, TTLSeconds: &ttl} + _, err := app.Submit(context.Background(), "alice", port.ConnInfo{}, req) + if protoCode(err) != protocol.CodeBadRequest { + t.Fatalf("ttl=0 want bad_request got %v", err) + } + }) + + t.Run("ttl_negative_rejected", func(t *testing.T) { + t.Parallel() + lim := defaultTestLimits() + app, db := openTestApp(t, lim) + insertEndpoint(t, db, "alice", "", 1, 0) + insertEndpoint(t, db, "bob", "", 1, 0) + ttl := int64(-1) + req := baseSend("ttlneg", "bob") + req.Offline = &protocol.OfflineOpts{Keep: true, TTLSeconds: &ttl} + _, err := app.Submit(context.Background(), "alice", port.ConnInfo{}, req) + if protoCode(err) != protocol.CodeBadRequest { + t.Fatalf("ttl=-1 want bad_request got %v", err) + } + }) + + t.Run("delay_maxint64_rejected", func(t *testing.T) { + t.Parallel() + lim := defaultTestLimits() + app, db := openTestApp(t, lim) + insertEndpoint(t, db, "alice", "", 1, 0) + insertEndpoint(t, db, "bob", "", 1, 0) + delay := int64(math.MaxInt64) + req := baseSend("delaymax", "bob") + req.DelayMs = &delay + _, err := app.Submit(context.Background(), "alice", port.ConnInfo{}, req) + if protoCode(err) != protocol.CodeBadRequest { + t.Fatalf("delay=MaxInt64 want bad_request got %v", err) + } + }) + + t.Run("sender_disabled_unauthorized", func(t *testing.T) { + t.Parallel() + lim := defaultTestLimits() + app, db := openTestApp(t, lim) + insertEndpoint(t, db, "alice", "", 0, 0) + insertEndpoint(t, db, "bob", "", 1, 0) + _, err := app.Submit(context.Background(), "alice", port.ConnInfo{}, baseSend("from-off", "bob")) + if protoCode(err) != protocol.CodeUnauthorized { + t.Fatalf("want unauthorized got %v", err) + } + var n int + if err := db.Read.QueryRow(`SELECT COUNT(*) FROM deliveries`).Scan(&n); err != nil { + t.Fatal(err) + } + if n != 0 { + t.Fatalf("deliveries=%d", n) + } + }) + + t.Run("group_late_joiner_skipped", func(t *testing.T) { + t.Parallel() + lim := defaultTestLimits() + app, db := openTestApp(t, lim) + insertEndpoint(t, db, "alice", "", 1, 0) + insertEndpoint(t, db, "bob", "", 1, 0) + insertEndpoint(t, db, "dave", "", 1, 0) + ctx := context.Background() + nowMs := int64(1_700_000_000_000) + err := db.Queue.Do(ctx, func(tx *sql.Tx) error { + if _, e := tx.Exec(`INSERT INTO groups(id, name, owner_id, created_at) VALUES(?,?,?,?)`, + "g-late", "g", "alice", nowMs); e != nil { + return e + } + for _, m := range []string{"alice", "bob"} { + if _, e := tx.Exec(`INSERT INTO group_members(group_id, endpoint_id, joined_at) VALUES(?,?,?)`, + "g-late", m, nowMs); e != nil { + return e + } + } + return nil + }) + if err != nil { + t.Fatal(err) + } + delay := int64(10_000) + req := &protocol.Send{ + V: protocol.Version, Type: protocol.TypeSend, RID: "r1", ID: "late-1", + To: protocol.Target{Kind: protocol.TargetGroup, ID: "g-late"}, + Body: protocol.Body{Enc: protocol.EncUTF8, Data: "hi"}, + DelayMs: &delay, + Offline: keepTrue(), + } + res, err := app.Submit(ctx, "alice", port.ConnInfo{}, req) + if err != nil { + t.Fatal(err) + } + if res.State != StateScheduled { + t.Fatalf("state=%s", res.State) + } + err = db.Queue.Do(ctx, func(tx *sql.Tx) error { + _, e := tx.Exec(`INSERT INTO group_members(group_id, endpoint_id, joined_at) VALUES(?,?,?)`, + "g-late", "dave", res.SendAtMs+500) + return e + }) + if err != nil { + t.Fatal(err) + } + if _, err := app.DispatchDue(ctx, res.SendAtMs, 10); err != nil { + t.Fatal(err) + } + var daveN int + if err := db.Read.QueryRow(` +SELECT COUNT(*) FROM deliveries d JOIN messages m ON m.seq=d.seq +WHERE m.id='late-1' AND d.endpoint_id='dave'`).Scan(&daveN); err != nil { + t.Fatal(err) + } + if daveN != 0 { + t.Fatalf("late joiner deliveries=%d", daveN) + } + var bobN int + if err := db.Read.QueryRow(` +SELECT COUNT(*) FROM deliveries d JOIN messages m ON m.seq=d.seq +WHERE m.id='late-1' AND d.endpoint_id='bob'`).Scan(&bobN); err != nil { + t.Fatal(err) + } + if bobN != 1 { + t.Fatalf("bob deliveries=%d", bobN) + } + }) + t.Run("rate_limited", func(t *testing.T) { t.Parallel() lim := defaultTestLimits() diff --git a/internal/app/message/void_finalize_test.go b/internal/app/message/void_finalize_test.go index 2fc75c3..5be2a6c 100644 --- a/internal/app/message/void_finalize_test.go +++ b/internal/app/message/void_finalize_test.go @@ -99,4 +99,47 @@ VALUES(?,?,?,0,'rejected','left_group',?)`, seq, "bob", nowMs, nowMs) if bodies != 0 { t.Fatalf("body still present: %d", bodies) } +} + +func TestCleanupKeepsRecentlyCompletedOldCreated(t *testing.T) { + t.Parallel() + lim := defaultTestLimits() + lim.RecordRetentionDays = 7 + app, db := openTestApp(t, lim) + insertEndpoint(t, db, "alice", "", 1, 0) + insertEndpoint(t, db, "bob", "", 1, 0) + ctx := context.Background() + nowMs := int64(1_700_000_000_000) + created := nowMs - int64(30)*24*3600*1000 + err := db.Queue.Do(ctx, func(tx *sql.Tx) error { + res, e := tx.Exec(` +INSERT INTO messages(id, sender_id, dest_kind, dest_id, meta, content_type, body_enc, + send_at, keep, ttl_seconds, receipt, state, reason, created_at) +VALUES('old-created','alice','endpoint','bob','{}','text/plain; charset=utf-8','utf8', + ?,0,0,1,'completed','',?)`, nowMs, created) + if e != nil { + return e + } + seq, e := res.LastInsertId() + if e != nil { + return e + } + _, e = tx.Exec(` +INSERT INTO deliveries(seq, endpoint_id, send_at, keep, state, reason, updated_at) +VALUES(?,?,?,0,'accepted','',?)`, seq, "bob", nowMs, nowMs) + return e + }) + if err != nil { + t.Fatal(err) + } + if err := app.CleanupOnce(ctx, nowMs); err != nil { + t.Fatal(err) + } + var n int + if err := db.Read.QueryRow(`SELECT COUNT(*) FROM messages WHERE id='old-created'`).Scan(&n); err != nil { + t.Fatal(err) + } + if n != 1 { + t.Fatalf("recently completed message should remain, n=%d", n) + } } \ No newline at end of file diff --git a/internal/protocol/types.go b/internal/protocol/types.go index 66d6f7c..68f8d89 100644 --- a/internal/protocol/types.go +++ b/internal/protocol/types.go @@ -23,9 +23,11 @@ type OfflineOpts struct { // Limits 是与服务器配置相关的校验上限。 type Limits struct { - MaxBodyBytes int - MaxMetaBytes int - MaxFrameBytes int + MaxBodyBytes int + MaxMetaBytes int + MaxFrameBytes int + MaxTTLSeconds int64 + MaxScheduleSeconds int64 } // DefaultLimits 返回 DEVELOPMENT 示例中的默认上限。 diff --git a/internal/protocol/validate.go b/internal/protocol/validate.go index 9d9785b..465db73 100644 --- a/internal/protocol/validate.go +++ b/internal/protocol/validate.go @@ -167,6 +167,23 @@ func (s *Send) Validate(lim Limits) error { if s.SendAtMs != nil && s.DelayMs != nil { return badRequest("send_at_ms and delay_ms are mutually exclusive") } + if EffectiveOfflineKeep(s) { + if s.Offline != nil && s.Offline.TTLSeconds != nil && *s.Offline.TTLSeconds <= 0 { + return badRequest("ttl_seconds must be > 0") + } + ttl := EffectiveOfflineTTL(s) + if lim.MaxTTLSeconds > 0 && ttl > lim.MaxTTLSeconds { + return badRequest("ttl_seconds exceeds max_ttl_seconds") + } + } + if s.DelayMs != nil { + if *s.DelayMs < 0 { + return badRequest("delay_ms negative") + } + if lim.MaxScheduleSeconds > 0 && *s.DelayMs > lim.MaxScheduleSeconds*1000 { + return badRequest("delay_ms exceeds max_schedule_seconds") + } + } n, err := FrameBytes(s) if err != nil { return badRequest("cannot encode frame")