feat(chat): 重试记忆的槽加 10 分钟时间上界 —— 别把用户新打的同内容当成重试
多槽让失败记录跨很多次发送存活，于是开了个单槽时代没有的窗口：用户几小时前失败过一句 「你好」，之后新打一遍「你好」（心里当的是一条新消息），指纹却与那条旧记录一字不差， keyFor 命中旧键 ⇒ 服务端回 duplicate ⇒ 这条新消息根本不发出去，屏幕上看着像「按了没反应」。 给每个槽加一道时间上界，窗口就被限死在 10 分钟内；而「刚失败完，用户马上再按一次发送」 这个真正需要幂等的场景一秒都没被削掉。 - RetryKeyMemory: 新增公开常量 SLOT_TTL_MILLIS = 10 分钟（测试拿它做断言，不硬编码）； RetryKeySlot 增 atMillis；remember(fingerprint, localId, atMillis) 与 keyFor(fingerprint, nowMillis, mint) 都在入口先按时间淘汰过期槽、再做自己那件事， 写入时最后才看条数上限（过期槽不占 8 条的额度）；同指纹重复失败更新键与时刻而不是留两条； forget(localId) 与时间无关，签名与行为不变；FIFO 上限 8 与 TTL 两条规则同时生效。 - 边界口径：正好等于 TTL 不算过期（判据是 now - atMillis > TTL，严格大于），理由写在类注释里， 测试把它钉住。 - ChatViewModel: 只改两个调用点，喂 System.currentTimeMillis()（emitAndDeliver 的 keyFor、 rememberRetryKey 的 remember；forget 的三处不动），并按新语义更新相关 KDoc。 - RetryKeyMemoryTest: 既有用例补上时刻；新增 7 条覆盖「过期槽不复用且被淘汰」「TTL 边界」 「remember 刷新时间戳」「过期的 A 不影响新鲜的 B」「TTL 先于 FIFO 上限生效」。
This commit is contained in:
AskaEth committed
1 parent
7a6f6b0863
commit
8938fffd08
3 files changed
+295
-66
No files matched your search
@@ -1,7 +1,16 @@
|
||||
package top.yeij.cyrene.domain.chat
|
||||
|
||||
/** 一次「没发出去」留下的记录：**那一条**的内容指纹 + 当时铸的幂等键（`localId`）。 */
|
||||
data class RetryKeySlot(val fingerprint: MessageFingerprint, val localId: String)
|
||||
/**
|
||||
* 一次「没发出去」留下的记录：**那一条**的内容指纹 + 当时铸的幂等键（`localId`）+ 记下它的时刻。
|
||||
*
|
||||
* [atMillis] 只服务于 [RetryKeyMemory.SLOT_TTL_MILLIS] 那道时间闸，与键本身的语义无关：没有它，一条
|
||||
* 几小时前的失败记录会一直活着，用户**新**打的一句同内容的话就会被当成重试（见 [RetryKeyMemory]）。
|
||||
*/
|
||||
data class RetryKeySlot(
|
||||
val fingerprint: MessageFingerprint,
|
||||
val localId: String,
|
||||
val atMillis: Long,
|
||||
)
|
||||
|
||||
/**
|
||||
* [RetryKeyMemory.keyFor] 的判定结果：这一次发送该用哪个幂等键，以及记忆变成什么样。
|
||||
@@ -12,7 +21,7 @@ data class RetryKeySlot(val fingerprint: MessageFingerprint, val localId: String
|
||||
data class RetryKeyDecision(val localId: String, val memory: RetryKeyMemory)
|
||||
|
||||
/**
|
||||
* 「没发出去的那几条」的记忆——**多槽**、纯值、可单测。
|
||||
* 「没发出去的那几条」的记忆——**多槽**、**有时间上界**、纯值、可单测。
|
||||
*
|
||||
* ## 它存在的理由
|
||||
* 幂等键要想真的挡住重复，就必须在**同一条消息**的两次发送之间保持不变。而「某次失败」并不
|
||||
@@ -22,6 +31,24 @@ data class RetryKeyDecision(val localId: String, val memory: RetryKeyMemory)
|
||||
* 让「一模一样的内容再发一次」落回同一个键，由服务端用 `chat_duplicate` /
|
||||
* `{duplicate:true,messageId}` 替本端裁决。
|
||||
*
|
||||
* ## 时间上界：10 分钟，为什么多槽必须配一个时钟
|
||||
* 记录是**跨很多次发送**存活的（这正是多槽的意义，见下面「为什么是多槽」），于是有了一个单槽时代
|
||||
* 不存在的窗口：用户**几小时前**失败过一句「你好」，之后**新**打一遍「你好」—— 他心里当的是一条
|
||||
* **新**消息，指纹却与那条旧记录一字不差 ⇒ [keyFor] 命中 ⇒ 复用旧键 ⇒ 服务端回 `chat_duplicate`
|
||||
* ⇒ 这条新消息**根本不发出去**，屏幕上看着像「按了没反应」。给每个槽加一道时间闸
|
||||
* [SLOT_TTL_MILLIS]，这个窗口就被钉死在 10 分钟内；而真正需要幂等的那一幕 ——「刚失败完，用户马上
|
||||
* 又按了一次发送」—— 一秒都没被削掉。
|
||||
*
|
||||
* 时钟由调用方喂（[remember] 的 `atMillis`、[keyFor] 的 `nowMillis`），纯值类型里**不藏**
|
||||
* `System.currentTimeMillis()`：判据才可单测，也才不用为了「等 10 分钟」写 sleep。
|
||||
*
|
||||
* ## 边界口径：正好等于 TTL **不算**过期
|
||||
* 判据是 `nowMillis - atMillis > SLOT_TTL_MILLIS`（严格大于），所以 `nowMillis == atMillis + TTL`
|
||||
* 那一条**仍然命中**，真正过期要再晚一毫秒。选这一边的理由：常量自身就能当「仍在窗口内」的边界值
|
||||
* 直接断言（测试不必为「还活着」那一侧发明魔法数），过期那一侧才需要显式的 `+1`；而真实时钟几乎
|
||||
* 不可能精确落在这 1 毫秒上，口径选哪边都没有实害，可测的那边更有价值。测试
|
||||
* `a slot exactly at the TTL boundary still counts as fresh` 把这条钉住。
|
||||
*
|
||||
* ## 三条纪律
|
||||
* 1. **只在回滚时写**（[remember]：发送失败 / 队列满被拒 / 附件与图片上传失败 / 图片处理失败…）：
|
||||
* 被受理的那条没有任何要记的东西。
|
||||
@@ -36,12 +63,19 @@ data class RetryKeyDecision(val localId: String, val memory: RetryKeyMemory)
|
||||
* 「服务端其实已经收下」的那条 B 落成**第二条**消息，正是本机制要消灭的重复。多槽让每条失败各留
|
||||
* 一个槽，用户重试哪一条就落回哪一条的键，与他的操作顺序无关。
|
||||
*
|
||||
* ## 上限与淘汰：FIFO，不是 LRU
|
||||
* ## 上限与淘汰：条数上限 8 + 时间上界 10 分钟，两条规则同时生效
|
||||
* 一条记录只在「用户可能还要重试」的窗口里有意义：失败后正文 / 图片 / 附件都还回了输入栏，他要么
|
||||
* 马上原样重发、要么早就改了内容（那就成了另一条消息，记录永远命中不了）。窗口外留下的只是垃圾，
|
||||
* 所以给一个小上界 [MAX_SLOTS]，满了丢**最旧**的（[remember] 里 FIFO）。刻意**不**做 LRU：命中
|
||||
* 并消费是**轮询**而不是访问 —— [keyFor] 把命中的那条整个摘走，与它有多旧无关，谈「最近使用」没有
|
||||
* 意义；也正因如此，剩下的槽天然还是写入顺序。
|
||||
* 所以有两把尺子：条数上限 [MAX_SLOTS]（满了丢**最旧**的，[remember] 里 FIFO）与时间上界
|
||||
* [SLOT_TTL_MILLIS]（比上限还旧的那条，无论排第几都作废）。
|
||||
*
|
||||
* **顺序是死的**：[remember] / [keyFor] **先按时间淘汰**过期槽，再做自己那件事（去重追加 / 查命中），
|
||||
* 写入时**最后**才看条数上限。先按时间清的好处是过期槽不占那份本该给新鲜记录的额度 —— 否则 8 个
|
||||
* 早已过期的槽会把刚写进来的那条挤出去（或者更糟：让 [keyFor] 在一个条数已满、内容却全过期的记忆里
|
||||
* 报「满了」）。
|
||||
*
|
||||
* 刻意**不**做 LRU：命中并消费是**轮询**而不是访问 —— [keyFor] 把命中的那条整个摘走，与它有多旧
|
||||
* 无关（只要求没过期），谈「最近使用」没有意义；也正因如此，剩下的槽天然还是写入顺序。
|
||||
*
|
||||
* ## 与「绝不自动重试」不矛盾
|
||||
* 这里一个字节都不自动重发：复用键只发生在**用户自己又按了一次发送**、且内容一字不差的时候。
|
||||
@@ -52,12 +86,20 @@ data class RetryKeyMemory(val slots: List<RetryKeySlot> = emptyList()) {
|
||||
/**
|
||||
* 记下「这一条没发出去」（回滚时调用）。
|
||||
*
|
||||
* 同一个指纹**先摘掉旧的、再把新的追加到末尾**：同一条又失败一次是**更新**它的键，而不是并列
|
||||
* 两条 —— 两条同指纹时 [keyFor] 只会命中更早那条，用户的重试反而用不上刚铸的键。超过
|
||||
* [MAX_SLOTS] 就从头部丢最旧的：本函数最多让**末尾**那条变新，淘汰因此绝不会伤到刚写的这条。
|
||||
* 三件事按固定顺序做，别调换（理由见类注释「上限与淘汰」）：
|
||||
* 1. **先按 [atMillis] 淘汰过期槽**——[atMillis] 既是这条记录的出生时刻，也是「现在」；
|
||||
* 2. 同一个指纹**先摘掉旧的、再把新的追加到末尾**：同一条又失败一次是**更新**它的键与时刻，
|
||||
* 而不是并列两条 —— 两条同指纹时 [keyFor] 只会命中更早那条，用户的重试反而用不上刚铸的键，
|
||||
* 而且那条更早的还会**先**过期（明明刚失败过，记录却已经超时，重试退回铸新键）；
|
||||
* 3. 超过 [MAX_SLOTS] 就从头部丢最旧的：本函数最多让**末尾**那条变新，淘汰因此绝不会伤到刚写的
|
||||
* 这条。
|
||||
*
|
||||
* @param atMillis 这次失败发生的时刻，由调用方喂（`System.currentTimeMillis()`）。
|
||||
*/
|
||||
fun remember(fingerprint: MessageFingerprint, localId: String): RetryKeyMemory {
|
||||
val kept = slots.filterNot { it.fingerprint == fingerprint } + RetryKeySlot(fingerprint, localId)
|
||||
fun remember(fingerprint: MessageFingerprint, localId: String, atMillis: Long): RetryKeyMemory {
|
||||
val fresh = purge(atMillis)
|
||||
val kept = fresh.slots.filterNot { it.fingerprint == fingerprint } +
|
||||
RetryKeySlot(fingerprint, localId, atMillis)
|
||||
return RetryKeyMemory(if (kept.size > MAX_SLOTS) kept.takeLast(MAX_SLOTS) else kept)
|
||||
}
|
||||
|
||||
@@ -67,6 +109,10 @@ data class RetryKeyMemory(val slots: List<RetryKeySlot> = emptyList()) {
|
||||
* 一条都对不上就**原样返回同一个实例**（一条也不清）：槽里躺的多半是别的刚回滚的消息，顺手清掉
|
||||
* 它们会让那几条的重试退回「铸新键」。同一个 `localId` 万一有多条（[remember] 保证不会有），
|
||||
* 一并清掉：留着一条键已被服务端受理过的记录，只会让将来的某次重发被当成重复吞掉。
|
||||
*
|
||||
* **与时间无关**：受理回执说的是「这条键服务端认了」，无论它记于几分钟前。此处也不顺手扫过期槽
|
||||
* —— 那要一个调用方没给的「现在」，而且过期槽本来就命中不了（[keyFor] 入口会清掉它们），留着
|
||||
* 最多是几个死条目的内存占用。
|
||||
*/
|
||||
fun forget(localId: String): RetryKeyMemory {
|
||||
val kept = slots.filterNot { it.localId == localId }
|
||||
@@ -74,25 +120,58 @@ data class RetryKeyMemory(val slots: List<RetryKeySlot> = emptyList()) {
|
||||
}
|
||||
|
||||
/**
|
||||
* 这一次发送该用哪个幂等键：内容与某条记录一致 ⇒ **复用**它的键，并**只消费那一条**（其余槽
|
||||
* 原样保留 —— 它们对应的是别的失败消息，正等着被各自的重试捡起来）；一条都对不上 ⇒ 铸一个新的
|
||||
* （[mint] 由调用方给，测试才能喂确定性的键，也才能断言「该不该铸」），**且一条槽也不清**。
|
||||
* 这一次发送该用哪个幂等键：内容与某条**未过期**的记录一致 ⇒ **复用**它的键，并**只消费那一条**
|
||||
* （其余槽原样保留 —— 它们对应的是别的失败消息，正等着被各自的重试捡起来）；一条都对不上 ⇒
|
||||
* 铸一个新的（[mint] 由调用方给，测试才能喂确定性的键，也才能断言「该不该铸」），**且一条槽也不清**。
|
||||
*
|
||||
* 入口**先**按 [nowMillis] 淘汰过期槽，过期的那条**视同不存在**：找不到就铸新键，与「从来没记过」
|
||||
* 这条完全一样。这正是时间上界要挡住的那一幕（见类注释「时间上界」）—— 几小时前失败过的同内容
|
||||
* 消息，今天再打一遍就是一条**新**消息，绝不能因为撞上旧键而被服务端当成重复吞掉。
|
||||
*
|
||||
* 最后这半句是与单槽时代最本质的差别，**别照着旧注释改回去**：那时记忆只有一条，「未命中就清空」
|
||||
* 与「保留」是同一件事；多槽下清空会把用户**别的**失败记录一并抹掉（他重试 A 时铸了新键，接着去
|
||||
* 重试 B，B 的槽已经没了 ⇒ 那次重发退化成非幂等）。未命中只说明「这次的正文没有任何失败记录对得
|
||||
* 上」，不是「记忆都作废了」。
|
||||
* 上」，不是「记忆都作废了」。（过期槽被摘掉不算例外：它们本来就再也命中不了。）
|
||||
*
|
||||
* @param nowMillis 这次发送的时刻，由调用方喂（`System.currentTimeMillis()`）。
|
||||
*/
|
||||
fun keyFor(fingerprint: MessageFingerprint, mint: () -> String): RetryKeyDecision {
|
||||
val hitIndex = slots.indexOfFirst { it.fingerprint == fingerprint }
|
||||
if (hitIndex < 0) return RetryKeyDecision(mint(), this)
|
||||
fun keyFor(
|
||||
fingerprint: MessageFingerprint,
|
||||
nowMillis: Long,
|
||||
mint: () -> String,
|
||||
): RetryKeyDecision {
|
||||
val fresh = purge(nowMillis)
|
||||
val hitIndex = fresh.slots.indexOfFirst { it.fingerprint == fingerprint }
|
||||
if (hitIndex < 0) return RetryKeyDecision(mint(), fresh)
|
||||
|
||||
val consumed = RetryKeyMemory(slots.filterIndexed { index, _ -> index != hitIndex })
|
||||
return RetryKeyDecision(slots[hitIndex].localId, consumed)
|
||||
val consumed = RetryKeyMemory(fresh.slots.filterIndexed { index, _ -> index != hitIndex })
|
||||
return RetryKeyDecision(fresh.slots[hitIndex].localId, consumed)
|
||||
}
|
||||
|
||||
/**
|
||||
* 摘掉在 [nowMillis] 这一刻已经过期的槽；一条都没过期就**原样返回同一个实例**（不必新建对象）。
|
||||
*
|
||||
* 过期判据见类注释「边界口径」——正好等于 [SLOT_TTL_MILLIS] 的那条**留着**。
|
||||
*/
|
||||
private fun purge(nowMillis: Long): RetryKeyMemory {
|
||||
val kept = slots.filterNot { it.isExpiredAt(nowMillis) }
|
||||
return if (kept.size == slots.size) this else RetryKeyMemory(kept)
|
||||
}
|
||||
|
||||
companion object {
|
||||
/** 同时保留的失败记录条数上限；淘汰策略见类注释「上限与淘汰」。 */
|
||||
const val MAX_SLOTS = 8
|
||||
|
||||
/**
|
||||
* 一条失败记录最多活多久（10 分钟）：超过它，[keyFor] 就再也认不出这条记录。
|
||||
*
|
||||
* 公开是因为**测试拿它做断言**（别在测试里重写一遍 `10 * 60 * 1000L`，那样改口径时测试不会
|
||||
* 跟着变）。为什么需要这道闸、边界怎么算，见类注释「时间上界」与「边界口径」。
|
||||
*/
|
||||
const val SLOT_TTL_MILLIS = 10 * 60 * 1000L
|
||||
|
||||
/** 这条记录在 [nowMillis] 这一刻是否已过期；`>` 而不是 `>=`，见类注释「边界口径」。 */
|
||||
private fun RetryKeySlot.isExpiredAt(nowMillis: Long): Boolean =
|
||||
nowMillis - atMillis > SLOT_TTL_MILLIS
|
||||
}
|
||||
}
|
||||
@@ -503,7 +503,9 @@ data class ChatUiState(
|
||||
* 幂等由 [MessageFingerprint] + [retryKey] 这一对承担：**回滚**时按内容指纹记下那条用过的
|
||||
* `clientMsgId`，用户**自己**再按一次发送、且内容一字不差时复用它（[emitAndDeliver] 是唯一的
|
||||
* 消费者）——服务端据此判重（`chat_duplicate` / `{duplicate:true}`）而不是落成第二条消息；
|
||||
* 内容改过（哪怕只多一个空格）就铸新键，因为那是另一条消息。**始终没有自动重试**：
|
||||
* 内容改过（哪怕只多一个空格）就铸新键，因为那是另一条消息。记录**不是永久的**：每条只活
|
||||
* [RetryKeyMemory.SLOT_TTL_MILLIS]（10 分钟，见那里「时间上界」）——多槽让记录跨很多次发送存活，
|
||||
* 太久以前那条失败过的内容再打一遍是**新**消息，不能被当成重试吞掉。**始终没有自动重试**：
|
||||
* 键的复用只发生在用户手动重发的那一刻。
|
||||
*
|
||||
* ## 单条消息的编辑 / 删除（D3）
|
||||
@@ -542,6 +544,12 @@ class ChatViewModel(
|
||||
* 刻意不清**：一条晚到的 `message_persisted` 恰恰是「服务端其实收下了、客户端当时判成失败」
|
||||
* 的证据，这时候把键留着，用户接下来的重试才会被判重而不是落成第二条消息 —— 那正是本机制
|
||||
* 存在的理由。
|
||||
*
|
||||
* 记录**有时间上界**（[RetryKeyMemory.SLOT_TTL_MILLIS] = 10 分钟）：比这还旧的槽在这一次发送时
|
||||
* 视同不存在。没有这道闸，多槽会让几小时前失败过的记录一直活着，用户**新**打一遍同样的内容就
|
||||
* 被服务端当成重复吞掉（看屏幕就是「按了没反应」）；而「刚失败完，用户马上再按一次发送」这个
|
||||
* 真正需要幂等的场景一秒都没被削掉。时间由上面两个调用点喂（`System.currentTimeMillis()`），
|
||||
* 这个纯值类型里不藏时钟。
|
||||
*/
|
||||
private var retryKey = RetryKeyMemory()
|
||||
|
||||
@@ -844,7 +852,12 @@ class ChatViewModel(
|
||||
// 复用键会让服务端判重并回 `chat_duplicate` / `{duplicate:true}`，而不是落成第二条消息。
|
||||
// 内容不同（改过字 / 换图 / 动了附件）则铸新键——用户的编辑是一条**新**消息。
|
||||
// 消费只发生在命中那一支：命中的那条记录在这里被摘走，别的失败记录原样留着（见 [RetryKeyMemory]）。
|
||||
val decision = retryKey.keyFor(MessageFingerprint.of(text, imageUri, documents)) {
|
||||
// 「现在」由这里喂进去（纯值类型里不藏时钟）：记下超过 `SLOT_TTL_MILLIS` 的那条已经不可复用、
|
||||
// 视同从来没记过——几小时前失败过同内容的消息，用户今天再打一遍是一条**新**消息。
|
||||
val decision = retryKey.keyFor(
|
||||
MessageFingerprint.of(text, imageUri, documents),
|
||||
System.currentTimeMillis(),
|
||||
) {
|
||||
"local-${UUID.randomUUID()}"
|
||||
}
|
||||
retryKey = decision.memory
|
||||
@@ -1172,7 +1185,10 @@ class ChatViewModel(
|
||||
}
|
||||
|
||||
/**
|
||||
* 把「这一条没发出去」记进 [retryKey]（指纹 + 当时用的幂等键）。
|
||||
* 把「这一条没发出去」记进 [retryKey]（指纹 + 当时用的幂等键 + 记下的**时刻**）。
|
||||
*
|
||||
* 时刻由这里喂（`System.currentTimeMillis()`），它是 [RetryKeyMemory.SLOT_TTL_MILLIS] 那道时间闸
|
||||
* 的起算点：同内容再次失败会把时刻一起刷新，所以「记得早」不会让一条刚更新的记录提前过期。
|
||||
*
|
||||
* 气泡**不在**待发列表里就什么都不记：那种情况下没有任何东西被回滚，记一条只会在下一次
|
||||
* 发送时凭空复用别人的键。传 [pending] 而不是自己读 `_state`，是因为调用点必须能在这条
|
||||
@@ -1183,6 +1199,7 @@ class ChatViewModel(
|
||||
retryKey = retryKey.remember(
|
||||
MessageFingerprint.of(failed.text, failed.imageUri, failed.documents),
|
||||
localId,
|
||||
System.currentTimeMillis(),
|
||||
)
|
||||
}
|
||||
|
||||
|
||||
Reference in new issue
Block a user