diff --git a/docs/DEVIATIONS.md b/docs/DEVIATIONS.md index 8c51f0c..b6eeac3 100644 --- a/docs/DEVIATIONS.md +++ b/docs/DEVIATIONS.md @@ -491,6 +491,15 @@ - 备选方案：总控在 `protocol` 增类型。 - 影响：接线编码 `resp.data` 时直接 Marshal 该 map 即可。 +### 复审修复 C-04 + +1. **退群/踢人/解散/停用/删除作废投递走统一终态函数** + - 原条款：DEVELOPMENT 7.6 投递进入 rejected 时写回执；没有 pending 时收尾 completed、删正文；保留 0 天同一事务删行。PRD F14/F18。 + - 实际做法：message 导出 `RejectPendingTx` / `TryFinalizeTx` / `FinalizeMessageTx`。group `void.go` 与 identity `lifecycle.go` 的作废/收尾改为调用它们，去掉复制 SQL。`sender_disabled`/`sender_deleted` 仍不写回执。`CleanupOnce` 分批收尾「dispatched 且无 pending」的卡住消息。作废路径未接线 `record_retention_days` 时按默认 7 天收尾（不在同一事务删行）；保留 0 天由 message 自己的 finalize 覆盖。不改 group `emit`。 + - 原因：原先 group 只改投递状态，identity 收尾但不写回执，最后一个 pending 被作废后消息永远停在 dispatched。 + - 备选方案：在 group/identity 各自补写回执与收尾（继续分叉）。 + - 影响：退群/解散/停用后发送方可收到 rejected 回执，配额释放，正文删除。 + ## 身份 I ### I1 2026-09-30 diff --git a/internal/app/group/group_test.go b/internal/app/group/group_test.go index cb18641..8fc0d6f 100644 --- a/internal/app/group/group_test.go +++ b/internal/app/group/group_test.go @@ -461,3 +461,62 @@ func TestGroupTransferRenameListGet(t *testing.T) { } } } + +func TestLeaveLastPendingFinalizesAndReceipt(t *testing.T) { + t.Parallel() + gApp, _, msgApp, db, _ := setup(t) + ctx := context.Background() + insertEP(t, db, "alice", 1) + insertEP(t, db, "bob", 1) + created, err := gApp.Create(ctx, "alice", &protocol.GroupCreate{ + V: protocol.Version, Type: protocol.TypeGroupCreate, RID: "1", + Name: "OnlyBob", Members: []protocol.GroupMemberIn{{ID: "bob"}}, + }) + if err != nil { + t.Fatal(err) + } + ttl := int64(3600) + _, err = msgApp.Submit(ctx, "alice", port.ConnInfo{}, &protocol.Send{ + V: protocol.Version, Type: protocol.TypeSend, RID: "s", ID: "keep1", + To: protocol.Target{Kind: protocol.TargetGroup, ID: created.ID}, + Body: protocol.Body{Enc: protocol.EncUTF8, Data: "hi"}, + Offline: &protocol.OfflineOpts{Keep: true, TTLSeconds: &ttl}, + }) + if err != nil { + t.Fatal(err) + } + if err = gApp.Leave(ctx, "bob", &protocol.GroupLeave{ + V: protocol.Version, Type: protocol.TypeGroupLeave, RID: "2", GroupID: created.ID, + }); err != nil { + t.Fatal(err) + } + var state, reason string + if err = db.Read.QueryRow(`SELECT state, reason FROM messages WHERE id='keep1'`).Scan(&state, &reason); err != nil { + t.Fatal(err) + } + if state != message.StateCompleted { + t.Fatalf("state=%s want completed", state) + } + var bodies int + if err = db.Read.QueryRow(`SELECT COUNT(*) FROM message_bodies b JOIN messages m ON m.seq=b.seq WHERE m.id='keep1'`).Scan(&bodies); err != nil { + t.Fatal(err) + } + if bodies != 0 { + t.Fatalf("body still present: %d", bodies) + } + var rState, rReason, rEP string + if err = db.Read.QueryRow(` +SELECT state, reason, endpoint_id FROM receipts WHERE sender_id='alice' AND msg_id='keep1'`).Scan(&rState, &rReason, &rEP); err != nil { + t.Fatalf("receipt: %v", err) + } + if rState != "rejected" || rReason != "left_group" || rEP != "bob" { + t.Fatalf("receipt state=%q reason=%q ep=%q", rState, rReason, rEP) + } + var pending int + if err = db.Read.QueryRow(`SELECT COUNT(*) FROM messages WHERE sender_id='alice' AND state IN ('scheduled','dispatched')`).Scan(&pending); err != nil { + t.Fatal(err) + } + if pending != 0 { + t.Fatalf("sender pending count=%d", pending) + } +} diff --git a/internal/app/group/void.go b/internal/app/group/void.go index 73d30c3..09e1e21 100644 --- a/internal/app/group/void.go +++ b/internal/app/group/void.go @@ -5,6 +5,7 @@ import ( "database/sql" "strings" + "git.asio.asia/nixevol/NixMsg/internal/app/message" "git.asio.asia/nixevol/NixMsg/internal/app/port" "git.asio.asia/nixevol/NixMsg/internal/protocol" ) @@ -19,7 +20,7 @@ type revokeItem struct { // voidMemberDeliveriesTx rejects pending deliveries for a leaving member; records revokes for pushed ones. func voidMemberDeliveriesTx(tx *sql.Tx, groupID, endpointID, reason string, nowMs int64, revokes *[]revokeItem) error { rows, err := tx.Query(` -SELECT d.seq, d.pushed_at, m.id, m.sender_id +SELECT d.seq, m.id, m.sender_id FROM deliveries d JOIN messages m ON m.seq = d.seq WHERE d.endpoint_id = ? AND d.state = 'pending' @@ -30,14 +31,13 @@ WHERE d.endpoint_id = ? AND d.state = 'pending' defer func() { _ = rows.Close() }() type row struct { seq int64 - pushed sql.NullInt64 msgID string senderID string } var list []row for rows.Next() { var r row - if scanErr := rows.Scan(&r.seq, &r.pushed, &r.msgID, &r.senderID); scanErr != nil { + if scanErr := rows.Scan(&r.seq, &r.msgID, &r.senderID); scanErr != nil { return scanErr } list = append(list, r) @@ -46,16 +46,18 @@ WHERE d.endpoint_id = ? AND d.state = 'pending' return err } for _, r := range list { - if _, execErr := tx.Exec(` -UPDATE deliveries SET state = 'rejected', reason = ?, updated_at = ? WHERE seq = ? AND endpoint_id = ? AND state = 'pending'`, - reason, nowMs, r.seq, endpointID); execErr != nil { + pushed, execErr := message.RejectPendingTx(tx, r.seq, endpointID, reason, nowMs) + if execErr != nil { return execErr } - if r.pushed.Valid && revokes != nil { + if pushed && revokes != nil { *revokes = append(*revokes, revokeItem{ endpointID: endpointID, msgID: r.msgID, fromID: r.senderID, reason: reason, }) } + if e := message.TryFinalizeTx(tx, r.seq, nowMs, message.DefaultVoidRetentionDays); e != nil { + return e + } } return nil } @@ -63,7 +65,7 @@ UPDATE deliveries SET state = 'rejected', reason = ?, updated_at = ? WHERE seq = // voidGroupAllTx rejects all pending group deliveries and completes scheduled messages. func voidGroupAllTx(tx *sql.Tx, groupID string, nowMs int64, revokes *[]revokeItem) error { rows, err := tx.Query(` -SELECT d.seq, d.endpoint_id, d.pushed_at, m.id, m.sender_id +SELECT d.seq, d.endpoint_id, m.id, m.sender_id FROM deliveries d JOIN messages m ON m.seq = d.seq WHERE d.state = 'pending' AND m.dest_kind = 'group' AND m.dest_id = ?`, groupID) @@ -73,14 +75,13 @@ WHERE d.state = 'pending' AND m.dest_kind = 'group' AND m.dest_id = ?`, groupID) type drow struct { seq int64 endpointID string - pushed sql.NullInt64 msgID string senderID string } var dlist []drow for rows.Next() { var r drow - if scanErr := rows.Scan(&r.seq, &r.endpointID, &r.pushed, &r.msgID, &r.senderID); scanErr != nil { + if scanErr := rows.Scan(&r.seq, &r.endpointID, &r.msgID, &r.senderID); scanErr != nil { _ = rows.Close() return scanErr } @@ -91,35 +92,35 @@ WHERE d.state = 'pending' AND m.dest_kind = 'group' AND m.dest_id = ?`, groupID) return err } for _, r := range dlist { - if _, execErr := tx.Exec(` -UPDATE deliveries SET state = 'rejected', reason = ?, updated_at = ? -WHERE seq = ? AND endpoint_id = ? AND state = 'pending'`, - reasonGroupDissolved, nowMs, r.seq, r.endpointID); execErr != nil { + pushed, execErr := message.RejectPendingTx(tx, r.seq, r.endpointID, reasonGroupDissolved, nowMs) + if execErr != nil { return execErr } - if r.pushed.Valid && revokes != nil { + if pushed && revokes != nil { *revokes = append(*revokes, revokeItem{ endpointID: r.endpointID, msgID: r.msgID, fromID: r.senderID, reason: reasonGroupDissolved, }) } + if e := message.TryFinalizeTx(tx, r.seq, nowMs, message.DefaultVoidRetentionDays); e != nil { + return e + } } srows, err := tx.Query(` -SELECT seq, id, sender_id, receipt FROM messages +SELECT seq, sender_id, receipt FROM messages WHERE dest_kind = 'group' AND dest_id = ? AND state = 'scheduled'`, groupID) if err != nil { return err } type srow struct { seq int64 - msgID string senderID string receipt int } var slist []srow for srows.Next() { var r srow - if scanErr := srows.Scan(&r.seq, &r.msgID, &r.senderID, &r.receipt); scanErr != nil { + if scanErr := srows.Scan(&r.seq, &r.senderID, &r.receipt); scanErr != nil { _ = srows.Close() return scanErr } @@ -130,23 +131,9 @@ WHERE dest_kind = 'group' AND dest_id = ? AND state = 'scheduled'`, groupID) return err } for _, r := range slist { - if _, execErr := tx.Exec(` -UPDATE messages SET state = 'completed', reason = ? WHERE seq = ? AND state = 'scheduled'`, - reasonGroupDissolved, r.seq); execErr != nil { + if execErr := message.FinalizeMessageTx(tx, r.seq, r.receipt != 0, r.senderID, "", reasonGroupDissolved, nowMs, message.DefaultVoidRetentionDays); execErr != nil { return execErr } - if _, execErr := tx.Exec(`DELETE FROM message_bodies WHERE seq = ?`, r.seq); execErr != nil { - return execErr - } - if r.receipt != 0 { - // 消息级作废回执：endpoint_id 空，state=rejected（DEVELOPMENT 6.4）；消息行仍为 completed - if _, execErr := tx.Exec(` -INSERT INTO receipts(sender_id, msg_id, endpoint_id, state, reason, created_at, acked) -VALUES(?,?,?,?,?,?,0)`, - r.senderID, r.msgID, "", "rejected", reasonGroupDissolved, nowMs); execErr != nil { - return execErr - } - } } return nil } diff --git a/internal/app/identity/lifecycle.go b/internal/app/identity/lifecycle.go index 422cae7..7c81177 100644 --- a/internal/app/identity/lifecycle.go +++ b/internal/app/identity/lifecycle.go @@ -7,6 +7,7 @@ import ( "errors" "time" + "git.asio.asia/nixevol/NixMsg/internal/app/message" "git.asio.asia/nixevol/NixMsg/internal/app/port" "git.asio.asia/nixevol/NixMsg/internal/protocol" ) @@ -140,9 +141,10 @@ WHERE id = ?`, endpointID); e != nil { } func voidEndpointMessagesTx(tx *sql.Tx, endpointID, recvReason, sendReason string, nowMs int64, revokes *[]revokeItem) error { + days := message.DefaultVoidRetentionDays // 发给 X 的 pending → rejected rows, err := tx.Query(` -SELECT d.seq, d.pushed_at, m.id, m.sender_id, m.receipt +SELECT d.seq, m.id, m.sender_id FROM deliveries d JOIN messages m ON m.seq = d.seq WHERE d.endpoint_id = ? AND d.state = 'pending'`, endpointID) @@ -151,15 +153,13 @@ WHERE d.endpoint_id = ? AND d.state = 'pending'`, endpointID) } type pendRow struct { seq int64 - pushed sql.NullInt64 msgID string senderID string - receipt int } var pending []pendRow for rows.Next() { var r pendRow - if scanErr := rows.Scan(&r.seq, &r.pushed, &r.msgID, &r.senderID, &r.receipt); scanErr != nil { + if scanErr := rows.Scan(&r.seq, &r.msgID, &r.senderID); scanErr != nil { _ = rows.Close() return scanErr } @@ -171,18 +171,11 @@ WHERE d.endpoint_id = ? AND d.state = 'pending'`, endpointID) } finalSeqs := map[int64]struct{}{} for _, r := range pending { - if _, execErr := tx.Exec(` -UPDATE deliveries SET state = 'rejected', reason = ?, updated_at = ? -WHERE seq = ? AND endpoint_id = ? AND state = 'pending'`, - recvReason, nowMs, r.seq, endpointID); execErr != nil { + pushed, execErr := message.RejectPendingTx(tx, r.seq, endpointID, recvReason, nowMs) + if execErr != nil { return execErr } - if r.receipt != 0 { - if e := insertReceiptIfWantedTx(tx, r.senderID, r.msgID, endpointID, "rejected", recvReason, nowMs, true); e != nil { - return e - } - } - if r.pushed.Valid && revokes != nil { + if pushed && revokes != nil { *revokes = append(*revokes, revokeItem{ endpointID: endpointID, msgID: r.msgID, fromID: r.senderID, reason: recvReason, }) @@ -192,21 +185,20 @@ WHERE seq = ? AND endpoint_id = ? AND state = 'pending'`, // 发给 X 的 scheduled 单聊 → completed，要回执则写 srows, err := tx.Query(` -SELECT seq, id, sender_id, receipt FROM messages +SELECT seq, sender_id, receipt FROM messages WHERE dest_kind = 'endpoint' AND dest_id = ? AND state = 'scheduled'`, endpointID) if err != nil { return err } type schedRow struct { seq int64 - msgID string senderID string receipt int } var scheduledTo []schedRow for srows.Next() { var r schedRow - if scanErr := srows.Scan(&r.seq, &r.msgID, &r.senderID, &r.receipt); scanErr != nil { + if scanErr := srows.Scan(&r.seq, &r.senderID, &r.receipt); scanErr != nil { _ = srows.Close() return scanErr } @@ -217,19 +209,8 @@ WHERE dest_kind = 'endpoint' AND dest_id = ? AND state = 'scheduled'`, endpointI return err } for _, r := range scheduledTo { - if _, execErr := tx.Exec(` -UPDATE messages SET state = 'completed', reason = ? WHERE seq = ? AND state = 'scheduled'`, - recvReason, r.seq); execErr != nil { - return execErr - } - if _, execErr := tx.Exec(`DELETE FROM message_bodies WHERE seq = ?`, r.seq); execErr != nil { - return execErr - } - if r.receipt != 0 { - // 消息级作废：endpoint_id 空，state=rejected（DEVELOPMENT 6.4） - if e := insertReceiptIfWantedTx(tx, r.senderID, r.msgID, "", "rejected", recvReason, nowMs, true); e != nil { - return e - } + if e := message.FinalizeMessageTx(tx, r.seq, r.receipt != 0, r.senderID, "", recvReason, nowMs, days); e != nil { + return e } } @@ -252,19 +233,14 @@ UPDATE messages SET state = 'completed', reason = ? WHERE seq = ? AND state = 's return err } for _, seq := range outSeqs { - if _, execErr := tx.Exec(` -UPDATE messages SET state = 'completed', reason = ? WHERE seq = ? AND state = 'scheduled'`, - sendReason, seq); execErr != nil { - return execErr - } - if _, execErr := tx.Exec(`DELETE FROM message_bodies WHERE seq = ?`, seq); execErr != nil { - return execErr + if e := message.FinalizeMessageTx(tx, seq, false, endpointID, "", sendReason, nowMs, days); e != nil { + return e } } // X 发出的消息的 pending 投递 → rejected（sender_*），不写回执 drows, err := tx.Query(` -SELECT d.seq, d.endpoint_id, d.pushed_at, m.id, m.sender_id +SELECT d.seq, d.endpoint_id, m.id, m.sender_id FROM deliveries d JOIN messages m ON m.seq = d.seq WHERE m.sender_id = ? AND d.state = 'pending'`, endpointID) @@ -274,14 +250,13 @@ WHERE m.sender_id = ? AND d.state = 'pending'`, endpointID) type outPend struct { seq int64 endpointID string - pushed sql.NullInt64 msgID string senderID string } var outPending []outPend for drows.Next() { var r outPend - if scanErr := drows.Scan(&r.seq, &r.endpointID, &r.pushed, &r.msgID, &r.senderID); scanErr != nil { + if scanErr := drows.Scan(&r.seq, &r.endpointID, &r.msgID, &r.senderID); scanErr != nil { _ = drows.Close() return scanErr } @@ -292,13 +267,11 @@ WHERE m.sender_id = ? AND d.state = 'pending'`, endpointID) return err } for _, r := range outPending { - if _, execErr := tx.Exec(` -UPDATE deliveries SET state = 'rejected', reason = ?, updated_at = ? -WHERE seq = ? AND endpoint_id = ? AND state = 'pending'`, - sendReason, nowMs, r.seq, r.endpointID); execErr != nil { + pushed, execErr := message.RejectPendingTx(tx, r.seq, r.endpointID, sendReason, nowMs) + if execErr != nil { return execErr } - if r.pushed.Valid && revokes != nil { + if pushed && revokes != nil { *revokes = append(*revokes, revokeItem{ endpointID: r.endpointID, msgID: r.msgID, fromID: r.senderID, reason: sendReason, }) @@ -307,7 +280,7 @@ WHERE seq = ? AND endpoint_id = ? AND state = 'pending'`, } for seq := range finalSeqs { - if e := tryFinalizeTx(tx, seq); e != nil { + if e := message.TryFinalizeTx(tx, seq, nowMs, days); e != nil { return e } } @@ -445,7 +418,7 @@ func withoutMember(ids []string, drop string) []string { // voidMemberDeliveriesTx 与 group 包同语义：退群成员的 pending 群投递改 rejected。 func voidMemberDeliveriesTx(tx *sql.Tx, groupID, endpointID, reason string, nowMs int64, revokes *[]revokeItem) error { rows, err := tx.Query(` -SELECT d.seq, d.pushed_at, m.id, m.sender_id +SELECT d.seq, m.id, m.sender_id FROM deliveries d JOIN messages m ON m.seq = d.seq WHERE d.endpoint_id = ? AND d.state = 'pending' @@ -455,14 +428,13 @@ WHERE d.endpoint_id = ? AND d.state = 'pending' } type row struct { seq int64 - pushed sql.NullInt64 msgID string senderID string } var list []row for rows.Next() { var r row - if scanErr := rows.Scan(&r.seq, &r.pushed, &r.msgID, &r.senderID); scanErr != nil { + if scanErr := rows.Scan(&r.seq, &r.msgID, &r.senderID); scanErr != nil { _ = rows.Close() return scanErr } @@ -472,19 +444,18 @@ WHERE d.endpoint_id = ? AND d.state = 'pending' if err = rows.Err(); err != nil { return err } + days := message.DefaultVoidRetentionDays for _, r := range list { - if _, execErr := tx.Exec(` -UPDATE deliveries SET state = 'rejected', reason = ?, updated_at = ? -WHERE seq = ? AND endpoint_id = ? AND state = 'pending'`, - reason, nowMs, r.seq, endpointID); execErr != nil { + pushed, execErr := message.RejectPendingTx(tx, r.seq, endpointID, reason, nowMs) + if execErr != nil { return execErr } - if r.pushed.Valid && revokes != nil { + if pushed && revokes != nil { *revokes = append(*revokes, revokeItem{ endpointID: endpointID, msgID: r.msgID, fromID: r.senderID, reason: reason, }) } - if e := tryFinalizeTx(tx, r.seq); e != nil { + if e := message.TryFinalizeTx(tx, r.seq, nowMs, days); e != nil { return e } } @@ -492,8 +463,9 @@ WHERE seq = ? AND endpoint_id = ? AND state = 'pending'`, } func voidGroupAllTx(tx *sql.Tx, groupID string, nowMs int64, revokes *[]revokeItem) error { + days := message.DefaultVoidRetentionDays rows, err := tx.Query(` -SELECT d.seq, d.endpoint_id, d.pushed_at, m.id, m.sender_id +SELECT d.seq, d.endpoint_id, m.id, m.sender_id FROM deliveries d JOIN messages m ON m.seq = d.seq WHERE d.state = 'pending' AND m.dest_kind = 'group' AND m.dest_id = ?`, groupID) @@ -503,14 +475,13 @@ WHERE d.state = 'pending' AND m.dest_kind = 'group' AND m.dest_id = ?`, groupID) type drow struct { seq int64 endpointID string - pushed sql.NullInt64 msgID string senderID string } var dlist []drow for rows.Next() { var r drow - if scanErr := rows.Scan(&r.seq, &r.endpointID, &r.pushed, &r.msgID, &r.senderID); scanErr != nil { + if scanErr := rows.Scan(&r.seq, &r.endpointID, &r.msgID, &r.senderID); scanErr != nil { _ = rows.Close() return scanErr } @@ -521,38 +492,35 @@ WHERE d.state = 'pending' AND m.dest_kind = 'group' AND m.dest_id = ?`, groupID) return err } for _, r := range dlist { - if _, execErr := tx.Exec(` -UPDATE deliveries SET state = 'rejected', reason = ?, updated_at = ? -WHERE seq = ? AND endpoint_id = ? AND state = 'pending'`, - reasonGroupDissolved, nowMs, r.seq, r.endpointID); execErr != nil { + pushed, execErr := message.RejectPendingTx(tx, r.seq, r.endpointID, reasonGroupDissolved, nowMs) + if execErr != nil { return execErr } - if r.pushed.Valid && revokes != nil { + if pushed && revokes != nil { *revokes = append(*revokes, revokeItem{ endpointID: r.endpointID, msgID: r.msgID, fromID: r.senderID, reason: reasonGroupDissolved, }) } - if e := tryFinalizeTx(tx, r.seq); e != nil { + if e := message.TryFinalizeTx(tx, r.seq, nowMs, days); e != nil { return e } } srows, err := tx.Query(` -SELECT seq, id, sender_id, receipt FROM messages +SELECT seq, sender_id, receipt FROM messages WHERE dest_kind = 'group' AND dest_id = ? AND state = 'scheduled'`, groupID) if err != nil { return err } type srow struct { seq int64 - msgID string senderID string receipt int } var slist []srow for srows.Next() { var r srow - if scanErr := srows.Scan(&r.seq, &r.msgID, &r.senderID, &r.receipt); scanErr != nil { + if scanErr := srows.Scan(&r.seq, &r.senderID, &r.receipt); scanErr != nil { _ = srows.Close() return scanErr } @@ -563,66 +531,13 @@ WHERE dest_kind = 'group' AND dest_id = ? AND state = 'scheduled'`, groupID) return err } for _, r := range slist { - if _, execErr := tx.Exec(` -UPDATE messages SET state = 'completed', reason = ? WHERE seq = ? AND state = 'scheduled'`, - reasonGroupDissolved, r.seq); execErr != nil { - return execErr - } - if _, execErr := tx.Exec(`DELETE FROM message_bodies WHERE seq = ?`, r.seq); execErr != nil { - return execErr - } - if r.receipt != 0 { - if e := insertReceiptIfWantedTx(tx, r.senderID, r.msgID, "", "rejected", reasonGroupDissolved, nowMs, true); e != nil { - return e - } + if e := message.FinalizeMessageTx(tx, r.seq, r.receipt != 0, r.senderID, "", reasonGroupDissolved, nowMs, days); e != nil { + return e } } return nil } -func insertReceiptIfWantedTx(tx *sql.Tx, senderID, msgID, endpointID, state, reason string, nowMs int64, alreadyWanted bool) error { - if !alreadyWanted { - return nil - } - var one int - err := tx.QueryRow(`SELECT 1 FROM endpoints WHERE id = ?`, senderID).Scan(&one) - if errors.Is(err, sql.ErrNoRows) { - return nil - } - if err != nil { - return err - } - _, err = tx.Exec(` -INSERT INTO receipts(sender_id, msg_id, endpoint_id, state, reason, created_at, acked) -VALUES(?,?,?,?,?,?,0)`, senderID, msgID, endpointID, state, reason, nowMs) - return err -} - -func tryFinalizeTx(tx *sql.Tx, seq int64) error { - var n int - if err := tx.QueryRow(`SELECT COUNT(*) FROM deliveries WHERE seq = ? AND state = 'pending'`, seq).Scan(&n); err != nil { - return err - } - if n > 0 { - return nil - } - var state string - if err := tx.QueryRow(`SELECT state FROM messages WHERE seq = ?`, seq).Scan(&state); err != nil { - if errors.Is(err, sql.ErrNoRows) { - return nil - } - return err - } - if state == "completed" { - return nil - } - if _, err := tx.Exec(`UPDATE messages SET state = 'completed' WHERE seq = ?`, seq); err != nil { - return err - } - _, err := tx.Exec(`DELETE FROM message_bodies WHERE seq = ?`, seq) - return err -} - func (a *App) publishRevokes(ctx context.Context, items []revokeItem) { if a.down == nil || len(items) == 0 { return diff --git a/internal/app/identity/lifecycle_test.go b/internal/app/identity/lifecycle_test.go index 951113f..f2d2a71 100644 --- a/internal/app/identity/lifecycle_test.go +++ b/internal/app/identity/lifecycle_test.go @@ -228,6 +228,54 @@ func TestF01DisableVoidsScheduledAndRejectsNew(t *testing.T) { } } +func TestDisableLastPendingFinalizesAndReceipt(t *testing.T) { + t.Parallel() + idApp, msgApp, db := openLifecycle(t) + ctx := context.Background() + insertEPFull(t, db, "alice") + insertEPFull(t, db, "bob") + ttl := int64(3600) + if _, err := msgApp.Submit(ctx, "alice", port.ConnInfo{EndpointID: "alice"}, &protocol.Send{ + V: protocol.Version, Type: protocol.TypeSend, RID: "1", ID: "keep-bob", + To: protocol.Target{Kind: protocol.TargetEndpoint, ID: "bob"}, + Body: protocol.Body{Enc: protocol.EncUTF8, Data: "hi"}, + Offline: &protocol.OfflineOpts{Keep: true, TTLSeconds: &ttl}, + }); err != nil { + t.Fatal(err) + } + if err := idApp.Disable(ctx, "bob"); err != nil { + t.Fatal(err) + } + var state, reason string + if err := db.Read.QueryRow(`SELECT state, reason FROM messages WHERE id='keep-bob'`).Scan(&state, &reason); err != nil { + t.Fatal(err) + } + if state != "completed" { + t.Fatalf("state=%s want completed", state) + } + var bodies int + if err := db.Read.QueryRow(`SELECT COUNT(*) FROM message_bodies b JOIN messages m ON m.seq=b.seq WHERE m.id='keep-bob'`).Scan(&bodies); err != nil { + t.Fatal(err) + } + if bodies != 0 { + t.Fatalf("body still present: %d", bodies) + } + var rState, rReason string + if err := db.Read.QueryRow(`SELECT state, reason FROM receipts WHERE sender_id='alice' AND msg_id='keep-bob'`).Scan(&rState, &rReason); err != nil { + t.Fatalf("receipt: %v", err) + } + if rState != "rejected" || rReason != "endpoint_disabled" { + t.Fatalf("receipt state=%q reason=%q", rState, rReason) + } + var pending int + if err := db.Read.QueryRow(`SELECT COUNT(*) FROM messages WHERE sender_id='alice' AND state IN ('scheduled','dispatched')`).Scan(&pending); err != nil { + t.Fatal(err) + } + if pending != 0 { + t.Fatalf("sender pending=%d", pending) + } +} + func TestF01DeleteOwnerTransfersEarliest(t *testing.T) { t.Parallel() idApp, _, db := openLifecycle(t) diff --git a/internal/app/message/ack.go b/internal/app/message/ack.go index 7c4c1b4..3085147 100644 --- a/internal/app/message/ack.go +++ b/internal/app/message/ack.go @@ -48,7 +48,7 @@ WHERE seq = ? AND endpoint_id = ? AND state = 'pending'`, if e := insertReceiptTx(tx, req.From, seq, endpointID, DeliveryAccepted, "", nowMs); e != nil { return e } - return tryFinalizeTx(tx, seq, nowMs, a.lim.RecordRetentionDays) + return TryFinalizeTx(tx, seq, nowMs, a.lim.RecordRetentionDays) } var state string err = tx.QueryRow(` @@ -175,7 +175,7 @@ SELECT COUNT(*) FROM deliveries WHERE seq = ? AND state IN ('expired','dropped', default: data.Result = "failed" } - return tryFinalizeTx(tx, seq, nowMs, a.lim.RecordRetentionDays) + return TryFinalizeTx(tx, seq, nowMs, a.lim.RecordRetentionDays) }) if err != nil { return data, err diff --git a/internal/app/message/dispatch.go b/internal/app/message/dispatch.go index 377ae8f..ac4c802 100644 --- a/internal/app/message/dispatch.go +++ b/internal/app/message/dispatch.go @@ -8,6 +8,9 @@ import ( "git.asio.asia/nixevol/NixMsg/internal/protocol" ) +// DefaultVoidRetentionDays 是 group/identity 作废路径未接线配置时的记录保留天数。 +const DefaultVoidRetentionDays = 7 + // 投递状态（DEVELOPMENT 7.1）。 const ( DeliveryPending = "pending" @@ -116,7 +119,7 @@ WHERE gm.group_id = ? AND gm.endpoint_id != ?`, destID, senderID) } if completeEarly { - if err := finalizeMessageTx(tx, seq, wantReceipt, senderID, "", msgReason, nowMs, a.lim.RecordRetentionDays); err != nil { + if err := FinalizeMessageTx(tx, seq, wantReceipt, senderID, "", msgReason, nowMs, a.lim.RecordRetentionDays); err != nil { return "", true, err } return StateCompleted, true, nil @@ -192,7 +195,7 @@ VALUES(?,?,?,?,?,?,?,NULL,NULL,0,?)`, } return StateDispatched, true, nil } - if err := finalizeMessageTx(tx, seq, wantReceipt, senderID, "", "", nowMs, a.lim.RecordRetentionDays); err != nil { + if err := FinalizeMessageTx(tx, seq, wantReceipt, senderID, "", "", nowMs, a.lim.RecordRetentionDays); err != nil { return "", true, err } return StateCompleted, true, nil @@ -212,9 +215,9 @@ func (a *App) lookupConn(endpointID string) (LiveConn, bool) { return a.conns.Current(endpointID) } -// finalizeMessageTx 无 pending 时收尾：completed、删正文；记录天数 0 则删消息与投递。 +// FinalizeMessageTx 无 pending 时收尾：completed、删正文；记录天数 0 则删消息与投递。 // msgReason 非空时写入消息 reason（发送前结束）；endpointID 为空表示消息级回执。 -func finalizeMessageTx(tx *sql.Tx, seq int64, wantReceipt bool, senderID, endpointID, msgReason string, nowMs int64, recordDays int) error { +func FinalizeMessageTx(tx *sql.Tx, seq int64, wantReceipt bool, senderID, endpointID, msgReason string, nowMs int64, recordDays int) error { var msgID string var receipt int if err := tx.QueryRow(`SELECT id, receipt FROM messages WHERE seq = ?`, seq).Scan(&msgID, &receipt); err != nil { @@ -244,8 +247,8 @@ func finalizeMessageTx(tx *sql.Tx, seq int64, wantReceipt bool, senderID, endpoi return nil } -// tryFinalizeTx 若无 pending 则收尾。 -func tryFinalizeTx(tx *sql.Tx, seq int64, nowMs int64, recordDays int) error { +// TryFinalizeTx 若无 pending 则收尾。 +func TryFinalizeTx(tx *sql.Tx, seq int64, nowMs int64, recordDays int) error { var n int if err := tx.QueryRow(`SELECT COUNT(*) FROM deliveries WHERE seq = ? AND state = 'pending'`, seq).Scan(&n); err != nil { return err @@ -261,7 +264,46 @@ func tryFinalizeTx(tx *sql.Tx, seq int64, nowMs int64, recordDays int) error { } return err } - return finalizeMessageTx(tx, seq, receipt != 0, senderID, "", "", nowMs, recordDays) + return FinalizeMessageTx(tx, seq, receipt != 0, senderID, "", "", nowMs, recordDays) +} + +func skipVoidReceipt(reason string) bool { + return reason == "sender_disabled" || reason == "sender_deleted" +} + +// RejectPendingTx 把一条 pending 投递改为 rejected；消息要求回执且发送方存在时写回执。 +// 停用/删除发送方（sender_disabled / sender_deleted）不写回执（DEVELOPMENT 7.6）。 +// 返回该投递是否曾推送，供调用方发 revoked。 +func RejectPendingTx(tx *sql.Tx, seq int64, endpointID, reason string, nowMs int64) (pushed bool, err error) { + var pushedAt sql.NullInt64 + err = tx.QueryRow(`SELECT pushed_at FROM deliveries WHERE seq = ? AND endpoint_id = ?`, seq, endpointID).Scan(&pushedAt) + if err == sql.ErrNoRows { + return false, nil + } + if err != nil { + return false, err + } + res, err := tx.Exec(` +UPDATE deliveries SET state = ?, reason = ?, updated_at = ? +WHERE seq = ? AND endpoint_id = ? AND state = ?`, + DeliveryRejected, reason, nowMs, seq, endpointID, DeliveryPending) + if err != nil { + return false, err + } + aff, _ := res.RowsAffected() + if aff == 0 { + return false, nil + } + if !skipVoidReceipt(reason) { + var senderID string + if err := tx.QueryRow(`SELECT sender_id FROM messages WHERE seq = ?`, seq).Scan(&senderID); err != nil { + return false, err + } + if err := insertReceiptTx(tx, senderID, seq, endpointID, DeliveryRejected, reason, nowMs); err != nil { + return false, err + } + } + return pushedAt.Valid, nil } func insertReceiptTx(tx *sql.Tx, senderID string, seq int64, endpointID, state, reason string, nowMs int64) error { diff --git a/internal/app/message/push.go b/internal/app/message/push.go index aa9d557..ec4a2cb 100644 --- a/internal/app/message/push.go +++ b/internal/app/message/push.go @@ -259,7 +259,7 @@ WHERE seq = ? AND endpoint_id = ? AND state = 'pending' AND pushed_conn IS NULL` if err := insertReceiptTx(tx, senderID, seq, endpointID, DeliveryRejected, ReasonTooLarge, nowMs); err != nil { return err } - return tryFinalizeTx(tx, seq, nowMs, a.lim.RecordRetentionDays) + return TryFinalizeTx(tx, seq, nowMs, a.lim.RecordRetentionDays) }) } @@ -361,7 +361,7 @@ WHERE seq = ? AND endpoint_id = ? AND state = 'pending'`, return err } } - if err := tryFinalizeTx(tx, seq, nowMs, a.lim.RecordRetentionDays); err != nil { + if err := TryFinalizeTx(tx, seq, nowMs, a.lim.RecordRetentionDays); err != nil { return err } if sendRevoked { diff --git a/internal/app/message/recover.go b/internal/app/message/recover.go index 3c9b64f..5edabc7 100644 --- a/internal/app/message/recover.go +++ b/internal/app/message/recover.go @@ -78,6 +78,10 @@ WHERE d.state = 'pending' AND d.pushed_conn IS NULL } } + if err := finalizeStuckDispatchedTx(tx, nowMs, a.lim.RecordRetentionDays); err != nil { + return err + } + if a.lim.RecordRetentionDays > 0 { cutoff := nowMs - int64(a.lim.RecordRetentionDays)*24*3600*1000 if _, err := tx.Exec(` @@ -120,3 +124,36 @@ DELETE FROM send_keys WHERE rowid IN ( a.flushRevokes(ctx) return nil } + +// finalizeStuckDispatchedTx 收尾「dispatched 且已无 pending 投递」的消息（C-04 兜底，修复已卡住的数据）。 +func finalizeStuckDispatchedTx(tx *sql.Tx, nowMs int64, recordDays int) error { + rows, err := tx.Query(` +SELECT seq FROM messages +WHERE state = ? + AND NOT EXISTS ( + SELECT 1 FROM deliveries d WHERE d.seq = messages.seq AND d.state = 'pending' + ) +LIMIT 500`, StateDispatched) + if err != nil { + return err + } + var seqs []int64 + for rows.Next() { + var seq int64 + if err := rows.Scan(&seq); err != nil { + _ = rows.Close() + return err + } + seqs = append(seqs, seq) + } + _ = rows.Close() + if err := rows.Err(); err != nil { + return err + } + for _, seq := range seqs { + if err := TryFinalizeTx(tx, seq, nowMs, recordDays); err != nil { + return err + } + } + return nil +} diff --git a/internal/app/message/void_finalize_test.go b/internal/app/message/void_finalize_test.go new file mode 100644 index 0000000..2fc75c3 --- /dev/null +++ b/internal/app/message/void_finalize_test.go @@ -0,0 +1,102 @@ +package message + +import ( + "context" + "database/sql" + "testing" + + "git.asio.asia/nixevol/NixMsg/internal/app/port" +) + +func TestRejectPendingAndFinalizeRetentionZero(t *testing.T) { + t.Parallel() + lim := defaultTestLimits() + lim.RecordRetentionDays = 0 + app, db := openTestApp(t, lim) + insertEndpoint(t, db, "alice", "", 1, 0) + insertEndpoint(t, db, "bob", "", 1, 0) + ctx := context.Background() + req := baseSend("z1", "bob") + req.Offline = keepTrue() + if _, err := app.Submit(ctx, "alice", port.ConnInfo{}, req); err != nil { + t.Fatal(err) + } + err := db.Queue.Do(ctx, func(tx *sql.Tx) error { + var seq int64 + if e := tx.QueryRow(`SELECT seq FROM messages WHERE id='z1'`).Scan(&seq); e != nil { + return e + } + if _, e := RejectPendingTx(tx, seq, "bob", ReasonEndpointDisabled, 1_700_000_000_000); e != nil { + return e + } + return TryFinalizeTx(tx, seq, 1_700_000_000_000, 0) + }) + if err != nil { + t.Fatal(err) + } + var n int + if err := db.Read.QueryRow(`SELECT COUNT(*) FROM messages WHERE id='z1'`).Scan(&n); err != nil { + t.Fatal(err) + } + if n != 0 { + t.Fatalf("message row should be deleted when retention=0, n=%d", n) + } + var receipts int + if err := db.Read.QueryRow(`SELECT COUNT(*) FROM receipts WHERE msg_id='z1' AND state='rejected'`).Scan(&receipts); err != nil { + t.Fatal(err) + } + if receipts != 1 { + t.Fatalf("receipts=%d", receipts) + } +} + +func TestCleanupStuckDispatched(t *testing.T) { + t.Parallel() + lim := defaultTestLimits() + 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) + 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('stuck','alice','endpoint','bob','{}','text/plain; charset=utf-8','utf8', + ?,0,0,1,'dispatched','',?)`, nowMs, nowMs) + if e != nil { + return e + } + seq, e := res.LastInsertId() + if e != nil { + return e + } + if _, e = tx.Exec(`INSERT INTO message_bodies(seq, body) VALUES(?, ?)`, seq, []byte("x")); e != nil { + return e + } + _, e = tx.Exec(` +INSERT INTO deliveries(seq, endpoint_id, send_at, keep, state, reason, updated_at) +VALUES(?,?,?,0,'rejected','left_group',?)`, seq, "bob", nowMs, nowMs) + return e + }) + if err != nil { + t.Fatal(err) + } + if err := app.CleanupOnce(ctx, nowMs); err != nil { + t.Fatal(err) + } + var state string + if err := db.Read.QueryRow(`SELECT state FROM messages WHERE id='stuck'`).Scan(&state); err != nil { + t.Fatal(err) + } + if state != StateCompleted { + t.Fatalf("state=%s want completed", state) + } + var bodies int + if err := db.Read.QueryRow(`SELECT COUNT(*) FROM message_bodies b JOIN messages m ON m.seq=b.seq WHERE m.id='stuck'`).Scan(&bodies); err != nil { + t.Fatal(err) + } + if bodies != 0 { + t.Fatalf("body still present: %d", bodies) + } +} \ No newline at end of file