nixevol/NixMsg
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

fix: 作废投递走统一终态函数并写回执收尾

Browse Source
This commit is contained in:
Nixevol
2026-09-30 14:58:58 +08:00
parent 4059a1576b
commit ac90495137
10 changed files with 365 additions and 166 deletions
Download Patch File Download Diff File Expand all files Collapse all files
docs/DEVIATIONS.md
+9
View File
@@ -448,6 +448,15 @@
- 备选方案：总控在 `protocol` 增类型。 - 备选方案：总控在 `protocol` 增类型。
- 影响：接线编码 `resp.data` 时直接 Marshal 该 map 即可。 - 影响：接线编码 `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 ## 身份 I
### I1 2026-09-30 ### I1 2026-09-30
internal/app/group/group_test.go
+59
View File
@@ -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)
}
}
internal/app/group/void.go
+20 -33
View File
@@ -5,6 +5,7 @@ import (
"database/sql" "database/sql"
"strings" "strings"
"git.asio.asia/nixevol/NixMsg/internal/app/message"
"git.asio.asia/nixevol/NixMsg/internal/app/port" "git.asio.asia/nixevol/NixMsg/internal/app/port"
"git.asio.asia/nixevol/NixMsg/internal/protocol" "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. // 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 { func voidMemberDeliveriesTx(tx *sql.Tx, groupID, endpointID, reason string, nowMs int64, revokes *[]revokeItem) error {
rows, err := tx.Query(` 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 FROM deliveries d
JOIN messages m ON m.seq = d.seq JOIN messages m ON m.seq = d.seq
WHERE d.endpoint_id = ? AND d.state = 'pending' WHERE d.endpoint_id = ? AND d.state = 'pending'
@@ -30,14 +31,13 @@ WHERE d.endpoint_id = ? AND d.state = 'pending'
defer func() { _ = rows.Close() }() defer func() { _ = rows.Close() }()
type row struct { type row struct {
seq int64 seq int64
pushed sql.NullInt64
msgID string msgID string
senderID string senderID string
} }
var list []row var list []row
for rows.Next() { for rows.Next() {
var r row 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 return scanErr
} }
list = append(list, r) list = append(list, r)
@@ -46,16 +46,18 @@ WHERE d.endpoint_id = ? AND d.state = 'pending'
return err return err
} }
for _, r := range list { for _, r := range list {
if _, execErr := tx.Exec(` pushed, execErr := message.RejectPendingTx(tx, r.seq, endpointID, reason, nowMs)
UPDATE deliveries SET state = 'rejected', reason = ?, updated_at = ? WHERE seq = ? AND endpoint_id = ? AND state = 'pending'`, if execErr != nil {
reason, nowMs, r.seq, endpointID); execErr != nil {
return execErr return execErr
} }
if r.pushed.Valid && revokes != nil { if pushed && revokes != nil {
*revokes = append(*revokes, revokeItem{ *revokes = append(*revokes, revokeItem{
endpointID: endpointID, msgID: r.msgID, fromID: r.senderID, reason: reason, 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 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. // voidGroupAllTx rejects all pending group deliveries and completes scheduled messages.
func voidGroupAllTx(tx *sql.Tx, groupID string, nowMs int64, revokes *[]revokeItem) error { func voidGroupAllTx(tx *sql.Tx, groupID string, nowMs int64, revokes *[]revokeItem) error {
rows, err := tx.Query(` 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 FROM deliveries d
JOIN messages m ON m.seq = d.seq JOIN messages m ON m.seq = d.seq
WHERE d.state = 'pending' AND m.dest_kind = 'group' AND m.dest_id = ?`, groupID) 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 { type drow struct {
seq int64 seq int64
endpointID string endpointID string
pushed sql.NullInt64
msgID string msgID string
senderID string senderID string
} }
var dlist []drow var dlist []drow
for rows.Next() { for rows.Next() {
var r drow 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() _ = rows.Close()
return scanErr return scanErr
} }
@@ -91,35 +92,35 @@ WHERE d.state = 'pending' AND m.dest_kind = 'group' AND m.dest_id = ?`, groupID)
return err return err
} }
for _, r := range dlist { for _, r := range dlist {
if _, execErr := tx.Exec(` pushed, execErr := message.RejectPendingTx(tx, r.seq, r.endpointID, reasonGroupDissolved, nowMs)
UPDATE deliveries SET state = 'rejected', reason = ?, updated_at = ? if execErr != nil {
WHERE seq = ? AND endpoint_id = ? AND state = 'pending'`,
reasonGroupDissolved, nowMs, r.seq, r.endpointID); execErr != nil {
return execErr return execErr
} }
if r.pushed.Valid && revokes != nil { if pushed && revokes != nil {
*revokes = append(*revokes, revokeItem{ *revokes = append(*revokes, revokeItem{
endpointID: r.endpointID, msgID: r.msgID, fromID: r.senderID, reason: reasonGroupDissolved, 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(` 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) WHERE dest_kind = 'group' AND dest_id = ? AND state = 'scheduled'`, groupID)
if err != nil { if err != nil {
return err return err
} }
type srow struct { type srow struct {
seq int64 seq int64
msgID string
senderID string senderID string
receipt int receipt int
} }
var slist []srow var slist []srow
for srows.Next() { for srows.Next() {
var r srow 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() _ = srows.Close()
return scanErr return scanErr
} }
@@ -130,23 +131,9 @@ WHERE dest_kind = 'group' AND dest_id = ? AND state = 'scheduled'`, groupID)
return err return err
} }
for _, r := range slist { for _, r := range slist {
if _, execErr := tx.Exec(` if execErr := message.FinalizeMessageTx(tx, r.seq, r.receipt != 0, r.senderID, "", reasonGroupDissolved, nowMs, message.DefaultVoidRetentionDays); execErr != nil {
UPDATE messages SET state = 'completed', reason = ? WHERE seq = ? AND state = 'scheduled'`,
reasonGroupDissolved, r.seq); execErr != nil {
return execErr 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 return nil
} }
internal/app/identity/lifecycle.go
+35 -120
View File
@@ -7,6 +7,7 @@ import (
"errors" "errors"
"time" "time"
"git.asio.asia/nixevol/NixMsg/internal/app/message"
"git.asio.asia/nixevol/NixMsg/internal/app/port" "git.asio.asia/nixevol/NixMsg/internal/app/port"
"git.asio.asia/nixevol/NixMsg/internal/protocol" "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 { func voidEndpointMessagesTx(tx *sql.Tx, endpointID, recvReason, sendReason string, nowMs int64, revokes *[]revokeItem) error {
days := message.DefaultVoidRetentionDays
// 发给 X 的 pending → rejected // 发给 X 的 pending → rejected
rows, err := tx.Query(` 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 FROM deliveries d
JOIN messages m ON m.seq = d.seq JOIN messages m ON m.seq = d.seq
WHERE d.endpoint_id = ? AND d.state = 'pending'`, endpointID) 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 { type pendRow struct {
seq int64 seq int64
pushed sql.NullInt64
msgID string msgID string
senderID string senderID string
receipt int
} }
var pending []pendRow var pending []pendRow
for rows.Next() { for rows.Next() {
var r pendRow 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() _ = rows.Close()
return scanErr return scanErr
} }
@@ -171,18 +171,11 @@ WHERE d.endpoint_id = ? AND d.state = 'pending'`, endpointID)
} }
finalSeqs := map[int64]struct{}{} finalSeqs := map[int64]struct{}{}
for _, r := range pending { for _, r := range pending {
if _, execErr := tx.Exec(` pushed, execErr := message.RejectPendingTx(tx, r.seq, endpointID, recvReason, nowMs)
UPDATE deliveries SET state = 'rejected', reason = ?, updated_at = ? if execErr != nil {
WHERE seq = ? AND endpoint_id = ? AND state = 'pending'`,
recvReason, nowMs, r.seq, endpointID); execErr != nil {
return execErr return execErr
} }
if r.receipt != 0 { if pushed && revokes != nil {
if e := insertReceiptIfWantedTx(tx, r.senderID, r.msgID, endpointID, "rejected", recvReason, nowMs, true); e != nil {
return e
}
}
if r.pushed.Valid && revokes != nil {
*revokes = append(*revokes, revokeItem{ *revokes = append(*revokes, revokeItem{
endpointID: endpointID, msgID: r.msgID, fromID: r.senderID, reason: recvReason, 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，要回执则写 // 发给 X 的 scheduled 单聊 → completed，要回执则写
srows, err := tx.Query(` 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) WHERE dest_kind = 'endpoint' AND dest_id = ? AND state = 'scheduled'`, endpointID)
if err != nil { if err != nil {
return err return err
} }
type schedRow struct { type schedRow struct {
seq int64 seq int64
msgID string
senderID string senderID string
receipt int receipt int
} }
var scheduledTo []schedRow var scheduledTo []schedRow
for srows.Next() { for srows.Next() {
var r schedRow 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() _ = srows.Close()
return scanErr return scanErr
} }
@@ -217,21 +209,10 @@ WHERE dest_kind = 'endpoint' AND dest_id = ? AND state = 'scheduled'`, endpointI
return err return err
} }
for _, r := range scheduledTo { for _, r := range scheduledTo {
if _, execErr := tx.Exec(` if e := message.FinalizeMessageTx(tx, r.seq, r.receipt != 0, r.senderID, "", recvReason, nowMs, days); e != nil {
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 return e
} }
} }
}
// X 发出的 scheduled → completed（sender_*），不写回执 // X 发出的 scheduled → completed（sender_*），不写回执
outSched, err := tx.Query(`SELECT seq FROM messages WHERE sender_id = ? AND state = 'scheduled'`, endpointID) outSched, err := tx.Query(`SELECT seq FROM messages WHERE sender_id = ? AND state = 'scheduled'`, endpointID)
@@ -252,19 +233,14 @@ UPDATE messages SET state = 'completed', reason = ? WHERE seq = ? AND state = 's
return err return err
} }
for _, seq := range outSeqs { for _, seq := range outSeqs {
if _, execErr := tx.Exec(` if e := message.FinalizeMessageTx(tx, seq, false, endpointID, "", sendReason, nowMs, days); e != nil {
UPDATE messages SET state = 'completed', reason = ? WHERE seq = ? AND state = 'scheduled'`, return e
sendReason, seq); execErr != nil {
return execErr
}
if _, execErr := tx.Exec(`DELETE FROM message_bodies WHERE seq = ?`, seq); execErr != nil {
return execErr
} }
} }
// X 发出的消息的 pending 投递 → rejected（sender_*），不写回执 // X 发出的消息的 pending 投递 → rejected（sender_*），不写回执
drows, err := tx.Query(` 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 FROM deliveries d
JOIN messages m ON m.seq = d.seq JOIN messages m ON m.seq = d.seq
WHERE m.sender_id = ? AND d.state = 'pending'`, endpointID) 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 { type outPend struct {
seq int64 seq int64
endpointID string endpointID string
pushed sql.NullInt64
msgID string msgID string
senderID string senderID string
} }
var outPending []outPend var outPending []outPend
for drows.Next() { for drows.Next() {
var r outPend 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() _ = drows.Close()
return scanErr return scanErr
} }
@@ -292,13 +267,11 @@ WHERE m.sender_id = ? AND d.state = 'pending'`, endpointID)
return err return err
} }
for _, r := range outPending { for _, r := range outPending {
if _, execErr := tx.Exec(` pushed, execErr := message.RejectPendingTx(tx, r.seq, r.endpointID, sendReason, nowMs)
UPDATE deliveries SET state = 'rejected', reason = ?, updated_at = ? if execErr != nil {
WHERE seq = ? AND endpoint_id = ? AND state = 'pending'`,
sendReason, nowMs, r.seq, r.endpointID); execErr != nil {
return execErr return execErr
} }
if r.pushed.Valid && revokes != nil { if pushed && revokes != nil {
*revokes = append(*revokes, revokeItem{ *revokes = append(*revokes, revokeItem{
endpointID: r.endpointID, msgID: r.msgID, fromID: r.senderID, reason: sendReason, 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 { for seq := range finalSeqs {
if e := tryFinalizeTx(tx, seq); e != nil { if e := message.TryFinalizeTx(tx, seq, nowMs, days); e != nil {
return e return e
} }
} }
@@ -445,7 +418,7 @@ func withoutMember(ids []string, drop string) []string {
// voidMemberDeliveriesTx 与 group 包同语义：退群成员的 pending 群投递改 rejected。 // voidMemberDeliveriesTx 与 group 包同语义：退群成员的 pending 群投递改 rejected。
func voidMemberDeliveriesTx(tx *sql.Tx, groupID, endpointID, reason string, nowMs int64, revokes *[]revokeItem) error { func voidMemberDeliveriesTx(tx *sql.Tx, groupID, endpointID, reason string, nowMs int64, revokes *[]revokeItem) error {
rows, err := tx.Query(` 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 FROM deliveries d
JOIN messages m ON m.seq = d.seq JOIN messages m ON m.seq = d.seq
WHERE d.endpoint_id = ? AND d.state = 'pending' WHERE d.endpoint_id = ? AND d.state = 'pending'
@@ -455,14 +428,13 @@ WHERE d.endpoint_id = ? AND d.state = 'pending'
} }
type row struct { type row struct {
seq int64 seq int64
pushed sql.NullInt64
msgID string msgID string
senderID string senderID string
} }
var list []row var list []row
for rows.Next() { for rows.Next() {
var r row 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() _ = rows.Close()
return scanErr return scanErr
} }
@@ -472,19 +444,18 @@ WHERE d.endpoint_id = ? AND d.state = 'pending'
if err = rows.Err(); err != nil { if err = rows.Err(); err != nil {
return err return err
} }
days := message.DefaultVoidRetentionDays
for _, r := range list { for _, r := range list {
if _, execErr := tx.Exec(` pushed, execErr := message.RejectPendingTx(tx, r.seq, endpointID, reason, nowMs)
UPDATE deliveries SET state = 'rejected', reason = ?, updated_at = ? if execErr != nil {
WHERE seq = ? AND endpoint_id = ? AND state = 'pending'`,
reason, nowMs, r.seq, endpointID); execErr != nil {
return execErr return execErr
} }
if r.pushed.Valid && revokes != nil { if pushed && revokes != nil {
*revokes = append(*revokes, revokeItem{ *revokes = append(*revokes, revokeItem{
endpointID: endpointID, msgID: r.msgID, fromID: r.senderID, reason: reason, 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 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 { func voidGroupAllTx(tx *sql.Tx, groupID string, nowMs int64, revokes *[]revokeItem) error {
days := message.DefaultVoidRetentionDays
rows, err := tx.Query(` 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 FROM deliveries d
JOIN messages m ON m.seq = d.seq JOIN messages m ON m.seq = d.seq
WHERE d.state = 'pending' AND m.dest_kind = 'group' AND m.dest_id = ?`, groupID) 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 { type drow struct {
seq int64 seq int64
endpointID string endpointID string
pushed sql.NullInt64
msgID string msgID string
senderID string senderID string
} }
var dlist []drow var dlist []drow
for rows.Next() { for rows.Next() {
var r drow 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() _ = rows.Close()
return scanErr return scanErr
} }
@@ -521,38 +492,35 @@ WHERE d.state = 'pending' AND m.dest_kind = 'group' AND m.dest_id = ?`, groupID)
return err return err
} }
for _, r := range dlist { for _, r := range dlist {
if _, execErr := tx.Exec(` pushed, execErr := message.RejectPendingTx(tx, r.seq, r.endpointID, reasonGroupDissolved, nowMs)
UPDATE deliveries SET state = 'rejected', reason = ?, updated_at = ? if execErr != nil {
WHERE seq = ? AND endpoint_id = ? AND state = 'pending'`,
reasonGroupDissolved, nowMs, r.seq, r.endpointID); execErr != nil {
return execErr return execErr
} }
if r.pushed.Valid && revokes != nil { if pushed && revokes != nil {
*revokes = append(*revokes, revokeItem{ *revokes = append(*revokes, revokeItem{
endpointID: r.endpointID, msgID: r.msgID, fromID: r.senderID, reason: reasonGroupDissolved, 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 return e
} }
} }
srows, err := tx.Query(` 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) WHERE dest_kind = 'group' AND dest_id = ? AND state = 'scheduled'`, groupID)
if err != nil { if err != nil {
return err return err
} }
type srow struct { type srow struct {
seq int64 seq int64
msgID string
senderID string senderID string
receipt int receipt int
} }
var slist []srow var slist []srow
for srows.Next() { for srows.Next() {
var r srow 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() _ = srows.Close()
return scanErr return scanErr
} }
@@ -563,66 +531,13 @@ WHERE dest_kind = 'group' AND dest_id = ? AND state = 'scheduled'`, groupID)
return err return err
} }
for _, r := range slist { for _, r := range slist {
if _, execErr := tx.Exec(` if e := message.FinalizeMessageTx(tx, r.seq, r.receipt != 0, r.senderID, "", reasonGroupDissolved, nowMs, days); e != nil {
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 return e
} }
} }
}
return nil 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) { func (a *App) publishRevokes(ctx context.Context, items []revokeItem) {
if a.down == nil || len(items) == 0 { if a.down == nil || len(items) == 0 {
return return
internal/app/identity/lifecycle_test.go
+48
View File
@@ -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) { func TestF01DeleteOwnerTransfersEarliest(t *testing.T) {
t.Parallel() t.Parallel()
idApp, _, db := openLifecycle(t) idApp, _, db := openLifecycle(t)
internal/app/message/ack.go
+2 -2
View File
@@ -48,7 +48,7 @@ WHERE seq = ? AND endpoint_id = ? AND state = 'pending'`,
if e := insertReceiptTx(tx, req.From, seq, endpointID, DeliveryAccepted, "", nowMs); e != nil { if e := insertReceiptTx(tx, req.From, seq, endpointID, DeliveryAccepted, "", nowMs); e != nil {
return e return e
} }
return tryFinalizeTx(tx, seq, nowMs, a.lim.RecordRetentionDays) return TryFinalizeTx(tx, seq, nowMs, a.lim.RecordRetentionDays)
} }
var state string var state string
err = tx.QueryRow(` err = tx.QueryRow(`
@@ -175,7 +175,7 @@ SELECT COUNT(*) FROM deliveries WHERE seq = ? AND state IN ('expired','dropped',
default: default:
data.Result = "failed" data.Result = "failed"
} }
return tryFinalizeTx(tx, seq, nowMs, a.lim.RecordRetentionDays) return TryFinalizeTx(tx, seq, nowMs, a.lim.RecordRetentionDays)
}) })
if err != nil { if err != nil {
return data, err return data, err
internal/app/message/dispatch.go
+49 -7
View File
@@ -8,6 +8,9 @@ import (
"git.asio.asia/nixevol/NixMsg/internal/protocol" "git.asio.asia/nixevol/NixMsg/internal/protocol"
) )
// DefaultVoidRetentionDays 是 group/identity 作废路径未接线配置时的记录保留天数。
const DefaultVoidRetentionDays = 7
// 投递状态（DEVELOPMENT 7.1）。 // 投递状态（DEVELOPMENT 7.1）。
const ( const (
DeliveryPending = "pending" DeliveryPending = "pending"
@@ -116,7 +119,7 @@ WHERE gm.group_id = ? AND gm.endpoint_id != ?`, destID, senderID)
} }
if completeEarly { 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 "", true, err
} }
return StateCompleted, true, nil return StateCompleted, true, nil
@@ -192,7 +195,7 @@ VALUES(?,?,?,?,?,?,?,NULL,NULL,0,?)`,
} }
return StateDispatched, true, nil 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 "", true, err
} }
return StateCompleted, true, nil return StateCompleted, true, nil
@@ -212,9 +215,9 @@ func (a *App) lookupConn(endpointID string) (LiveConn, bool) {
return a.conns.Current(endpointID) return a.conns.Current(endpointID)
} }
// finalizeMessageTx 无 pending 时收尾：completed、删正文；记录天数 0 则删消息与投递。 // FinalizeMessageTx 无 pending 时收尾：completed、删正文；记录天数 0 则删消息与投递。
// msgReason 非空时写入消息 reason（发送前结束）；endpointID 为空表示消息级回执。 // 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 msgID string
var receipt int var receipt int
if err := tx.QueryRow(`SELECT id, receipt FROM messages WHERE seq = ?`, seq).Scan(&msgID, &receipt); err != nil { 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 return nil
} }
// tryFinalizeTx 若无 pending 则收尾。 // TryFinalizeTx 若无 pending 则收尾。
func tryFinalizeTx(tx *sql.Tx, seq int64, nowMs int64, recordDays int) error { func TryFinalizeTx(tx *sql.Tx, seq int64, nowMs int64, recordDays int) error {
var n int var n int
if err := tx.QueryRow(`SELECT COUNT(*) FROM deliveries WHERE seq = ? AND state = 'pending'`, seq).Scan(&n); err != nil { if err := tx.QueryRow(`SELECT COUNT(*) FROM deliveries WHERE seq = ? AND state = 'pending'`, seq).Scan(&n); err != nil {
return err return err
@@ -261,7 +264,46 @@ func tryFinalizeTx(tx *sql.Tx, seq int64, nowMs int64, recordDays int) error {
} }
return err 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 { func insertReceiptTx(tx *sql.Tx, senderID string, seq int64, endpointID, state, reason string, nowMs int64) error {
internal/app/message/push.go
+2 -2
View File
@@ -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 { if err := insertReceiptTx(tx, senderID, seq, endpointID, DeliveryRejected, ReasonTooLarge, nowMs); err != nil {
return err 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 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 return err
} }
if sendRevoked { if sendRevoked {
internal/app/message/recover.go
+37
View File
@@ -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 { if a.lim.RecordRetentionDays > 0 {
cutoff := nowMs - int64(a.lim.RecordRetentionDays)*24*3600*1000 cutoff := nowMs - int64(a.lim.RecordRetentionDays)*24*3600*1000
if _, err := tx.Exec(` if _, err := tx.Exec(`
@@ -120,3 +124,36 @@ DELETE FROM send_keys WHERE rowid IN (
a.flushRevokes(ctx) a.flushRevokes(ctx)
return nil 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
}
internal/app/message/void_finalize_test.go
+102
View File
@@ -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)
}
}