W77 增量追尾:满页之后的空页不再当作尾巴,未读基数的边界如实登记

review 判了两条 P2,都只动 W77 这一摊。

1) `loadIncrement` 里 `newest <= cursor` 那条同毫秒防线是死代码。服务端的 `after` 是
**严格** `>`(`list_after.go` 的 `ListAfter` 与 `devtools/fake-backend` 一致),返回的
每一行都 `createdAt > cursor`,那个分支永不触发。真实风险在别处:满页恰好把同一毫秒劈开
时,下一次 `after=newest` 只会回空页,旧代码把那次空页当成「追到了尾巴」⇒ 静默少给几条
且不回落。改成「拿过一整页之后紧跟的空页 ⇒ `complete = false`」:客户端分不出干净尾巴与
这个洞,就如实说没追完,把决定权交回调用方(未读面上是回落整表)。第一页就空仍算干净
尾巴(没有可少给的),不满一页仍算追到尾巴,翻满预算仍算没追完。

连带把测试改成真后端的形状:满页后空页那条的期望从 `complete = true` 改成 `false`;同
毫秒那条不再让假后端把同一批行回两次(严格 `>` 下后端给不出这种响应),改成「三条同毫秒、
页宽 2、第二页空」并断言第二次请求真的带 `after=500`;新增「第一页空页 ⇒ `complete =
true`」把新逻辑的另一半钉住。KDoc 里「再翻也只会拿到同一批行」这类与契约矛盾的措辞一并
重写。

2) `applyIncrement` 的 `base + unseen` 在断线期间他端已读时会多计:`session_read` 帧永久
丢,`read_at` 又是账号级游标。代码行为不动,只把措辞限定清楚 ——「同源」的只是**过滤口径**
(`count(created_at > read_at)`),基数仍是上次整表的旧值,跨端读会让它漂移,由下一次整表
刷新自愈。这是**已接受的边界**,KDoc 写明了,另加一条测试把它钉住
(`a read elsewhere while offline over-counts until the next whole table`),免得日后被
当成「已修好」。

不装「`readAt > from` ⇒ 回落整表」的保守闸:本端刚点开过该会话时 `readAt` 本来就被推到
「现在」、必然大于 `lastMessageAt`,那是最常见的一档;拿它当异常信号会让增量路径在正常
使用下持续回落整表,而真正的「他端已读」它一样抓不全(断线时帧就丢了)。抓不准的闸门不如
不装,理由写在 KDoc 里。

顺手把 `SessionItem.lastMessageAt` 与 `flushPendingIncrements` 里过时的注释改成新口径
(它现在同时是增量追尾的 `after` 游标)。

验证:`./gradlew.bat testDebugUnitTest assembleDebug` 全绿 —— 135 个 suite / 1691 条
(比上轮 +2)/ 0 失败 / 0 跳过,`assembleDebug` 打包成功。
This commit is contained in:
AskaEth committed 2026-10-26 23:21:15 +08:00
1 parent d716d26864
commit db53de1a7e
5 files changed
+101 -28

No files matched your search

@@ -88,7 +88,9 @@ class MessageRepository(
* `createdAt`([PagedMessages.newestCreatedAt]),不是最旧的。 * `createdAt`([PagedMessages.newestCreatedAt]),不是最旧的。
* *
* 同毫秒并列且条数超过 `limit` 时,纯 `createdAt` 游标推不动(契约 §A5 明说的已知边界)—— * 同毫秒并列且条数超过 `limit` 时,纯 `createdAt` 游标推不动(契约 §A5 明说的已知边界)——
* 调用方按 `id` 去重、并在游标没前进时停下来,不在这里兜。 * 落在页边界上的那一毫秒里剩下的行**再也追不回来**,因为服务端只给严格 `>`。调用方按 `id`
* 去重,并且**不能把「满页之后的空页」当作追到了尾巴**:要翻页又要把边界说清楚,用
* [loadIncrement],不要自己拿这一页凑。
*/ */
suspend fun loadAfter( suspend fun loadAfter(
sessionId: String, sessionId: String,
@@ -104,10 +106,24 @@ class MessageRepository(
/** /**
* **增量追尾**(W77):从 [after] 起把某会话的新消息一直翻到尾巴,最多 [maxPages] 页。 * **增量追尾**(W77):从 [after] 起把某会话的新消息一直翻到尾巴,最多 [maxPages] 页。
* *
* 与裸的 [loadAfter] 的差别只有一件:**它负责把「还没翻完」这件事如实说出来**。`after` 方向的 * 与裸的 [loadAfter] 的差别只有一件:**它负责把「还没翻完」这件事如实说出来**。
* 游标是本页**最新**一条的 `createdAt`,同毫秒并列超过一页时纯 `createdAt` 游标推不动 *
* (契约 §A5 明说的已知边界),再翻也只会拿到同一批行——那种情况返回 `complete = false`, * ## 什么时候只能说 `complete = false`
* 由调用方决定怎么办(未读面上是**回落到整表**),这里**不猜**、也不悄悄少给几条。 * 游标是本页**最新**一条的 `createdAt`,而服务端只给 `created_at > after`(**严格**大于:
* `list_after.go` 的 `ListAfter` 与 `devtools/fake-backend` 两处一致)。于是**落在页边界上被
* 劈开的那一毫秒**里的行永远追不回来——它们的 `createdAt` 恰好等于游标,再问一次只会得到空页
* (契约 §A5 的已知边界:纯 `createdAt` 游标没有毫秒内的次序)。
*
* 客户端**分不出**那次空页是「干净的尾巴」还是「空页底下还压着同一毫秒的几条」,所以:
*
* - **拿过一整页之后紧跟的空页 ⇒ `complete = false`**,把决定权交回调用方(未读面上是回落到
* 整表)。代价是「积压条数正好是页宽的整数倍」这种干净尾巴也要回落一次——罕见,且比静默少给
* 几条便宜得多;
* - **第一页就是空页 ⇒ `complete = true`**:一条新消息都没有,没有可少给的;
* - **不满一页 ⇒ `complete = true`**:服务端 `LIMIT` 语义,给不满就说明 `> cursor` 的给完了;
* - **翻满 [maxPages] ⇒ `complete = false`**。
*
* 这里**不猜**、也不悄悄少给几条:说不完整,调用方还能去问唯一权威。
* *
* 不传 `platform`:未读面上的口径是服务端 `count(created_at > read_at)`(`unreadCountQuery`, * 不传 `platform`:未读面上的口径是服务端 `count(created_at > read_at)`(`unreadCountQuery`,
* 不带平台过滤),按平台过滤会让本地数出来的条数**与服务端的 `unreadCount` 不同源**。 * 不带平台过滤),按平台过滤会让本地数出来的条数**与服务端的 `unreadCount` 不同源**。
@@ -126,6 +142,8 @@ class MessageRepository(
val size = pageSize.coerceIn(1, MAX_PAGE_SIZE) val size = pageSize.coerceIn(1, MAX_PAGE_SIZE)
val byId = LinkedHashMap<String, HistoryMessageDto>() val byId = LinkedHashMap<String, HistoryMessageDto>()
var cursor = after var cursor = after
// 已经翻到过一整页 ⇒ 此后那个空页不再可信(页边界可能劈开了一毫秒,见 KDoc)。
var chasedFullPage = false
fun done(at: Long, complete: Boolean) = Result.success( fun done(at: Long, complete: Boolean) = Result.success(
MessageIncrement(messages = byId.values.sortedWith(MessageOrder), cursor = at, complete = complete), MessageIncrement(messages = byId.values.sortedWith(MessageOrder), cursor = at, complete = complete),
@@ -137,14 +155,17 @@ class MessageRepository(
val raw = apiCall { api.listMessages(sessionId, size, before = null, platform = platform, after = cursor) } val raw = apiCall { api.listMessages(sessionId, size, before = null, platform = platform, after = cursor) }
.getOrElse { return@withContext Result.failure(it) } .getOrElse { return@withContext Result.failure(it) }
val paged = toPagedMessages(raw, size) val paged = toPagedMessages(raw, size)
// 空页 = 真的追到了尾巴。`hasMore` 只是「拿到满页」的保守判断,条数正好是页宽的整数倍时 // 空页有两种含义,客户端分不出来:第一页就空 = 真的没有新的;拿过一整页之后空 = 可能还有
// 会多问一次,那一次拿到的就是这一页空的。 // 同一毫秒的几条卡在游标上(服务端只给严格大于)。后者只能说「没追完」。
if (paged.messages.isEmpty()) return@withContext done(cursor, complete = true) if (paged.messages.isEmpty()) {
return@withContext done(cursor, complete = !chasedFullPage)
}
// 去重只是防御:严格 `>` 下相邻两页不可能重叠(下一页只要 > 本页最新那条)。
paged.messages.forEach { byId[it.id] = it } paged.messages.forEach { byId[it.id] = it }
val newest = paged.newestCreatedAt ?: return@withContext done(cursor, complete = false) val newest = paged.newestCreatedAt ?: return@withContext done(cursor, complete = false)
if (!paged.hasMore) return@withContext done(newest, complete = true) if (!paged.hasMore) return@withContext done(newest, complete = true)
if (newest <= cursor) return@withContext done(newest, complete = false)
cursor = newest cursor = newest
chasedFullPage = true
} }
done(cursor, complete = false) done(cursor, complete = false)
} }
@@ -252,14 +252,26 @@ internal fun incrementCursors(
* *
* ## 未读数怎么算:先过已读游标,再加 * ## 未读数怎么算:先过已读游标,再加
* 增量条数**不是**直接往角标上加的,要先按本地已知的已读游标 [readAt](`readCursors[sessionId]`)筛一遍。 * 增量条数**不是**直接往角标上加的,要先按本地已知的已读游标 [readAt](`readCursors[sessionId]`)筛一遍。
* 服务端的口径是 `count(created_at > read_at)`(`session_read.go` 的 `unreadCountQuery`),而 `read_at` * **过滤口径**与服务端同源:`session_read.go` 的 `unreadCountQuery` 是 `count(created_at > read_at)`,
* 是**账号级单一游标**,可能比本端已知的最新消息还新——用户在这台机器上点开过它([SessionsViewModel.select] * 而 `read_at` 是**账号级单一游标**,可能比本端已知的最新消息还新——用户在这台机器上点开过它
* 推的就是「现在」),随后别的设备又发了消息。那一刻 `(lastMessageAt, readAt]` 之间的消息**服务端算已读**, * ([SessionsViewModel.select] 推的就是「现在」),随后别的设备又发了消息。那一刻
* 一律加进角标就会多出来一个红点。所以只数 `createdAt > readAt` 的那几条。 * `(lastMessageAt, readAt]` 之间的消息**服务端算已读**,一律加进角标就会多出来一个红点。所以只数
* `createdAt > readAt` 的那几条。
* *
* 已有的 [SessionItem.unreadCount] 直接保留相加是安全的:它来自服务端,而这次增量里的每一条都严格新于 * 已有的 [SessionItem.unreadCount] 直接保留相加是安全的:它来自服务端,而这次增量里的每一条都严格新于
* [from](服务端的 `after` 就是 `created_at > ?`),两者没有交集,不会重复计数。 * [from](服务端的 `after` 就是 `created_at > ?`),两者没有交集,不会重复计数。
* *
* ## 已知边界:同源的只是**过滤口径**,基数仍是上次整表的旧值
* 相加的底数 [SessionItem.unreadCount] 来自上一次整表刷新,而 `read_at` 是账号级的、会被别的设备改写。
* 断线期间他端把某个会话读掉时,那条 `session_read` 帧本地**永久收不到**(断线不补帧),于是底数偏大,
* 这一行会一直多算那几条。这是**已接受的边界**,不是漏网:红点本来就只承诺「大概有几条新的」,漂移
* 到下一次整表刷新(打开抽屉 / 初始化 / 任何一次回落)就自愈。
*
* 这里**不**去检测它(例如「[readAt] > [from] ⇒ 底数可疑 ⇒ 回落整表」):本端刚点开过这个会话时
* `readAt` 本来就被推到「现在」、必然大于 `lastMessageAt`,那是最常见的一档;拿它当异常信号会让增量
* 路径在正常使用下持续回落整表,而真正的「他端已读」它一样抓不全(断线时帧就丢了)。抓不准的闸门
* 不如不装。
*
* ## 并发的那一份真相 * ## 并发的那一份真相
* [from] 不是「这次用了什么游标」的装饰:行上的 [SessionItem.lastMessageAt] **已经不等于** [from], * [from] 不是「这次用了什么游标」的装饰:行上的 [SessionItem.lastMessageAt] **已经不等于** [from],
* 就说明在这次请求飞在天上的时候,另一份更权威的结果(一次整表刷新、或另一次增量)已经推进过这一行。 * 就说明在这次请求飞在天上的时候,另一份更权威的结果(一次整表刷新、或另一次增量)已经推进过这一行。
@@ -28,7 +28,8 @@ data class SessionItem(
val lastMessage: String?, val lastMessage: String?,
/** /**
* 最后一条消息的时间戳(毫秒);会话还没有消息时为 null。 * 最后一条消息的时间戳(毫秒);会话还没有消息时为 null。
* 只用于已读游标比对(见 [SessionItem.withReadCursor]),不直接显示。 * 两处用:已读游标比对(见 [SessionItem.withReadCursor]),以及增量追尾的 `after` 游标
* (见 `incrementCursorFor`,由 `applyIncrement` 推进到这次增量的 `cursor`)。不直接显示。
*/ */
val lastMessageAt: Long? = null, val lastMessageAt: Long? = null,
/** 未读条数(服务端 `unreadCount`,D4);`0` = 无未读、不显示角标。 */ /** 未读条数(服务端 `unreadCount`,D4);`0` = 无未读、不显示角标。 */
@@ -412,9 +413,10 @@ class SessionsViewModel(private val container: AppContainer) : ViewModel() {
* *
* ## 回落的三条理由(都落到同一条路:整表) * ## 回落的三条理由(都落到同一条路:整表)
* 1. **问不了增量**:[incrementCursors] 给 `null` —— 行不在列表里,或本端不知道它看到哪儿了; * 1. **问不了增量**:[incrementCursors] 给 `null` —— 行不在列表里,或本端不知道它看到哪儿了;
* 2. **追不到尾**:服务端说有更多却推不动(`complete = false`:同毫秒并列超过一页、或翻满了 * 2. **追不到尾**:`complete = false` —— 拿过一整页之后紧跟的空页(页边界可能劈开了同一毫秒,
* `MessageRepository.MAX_INCREMENT_PAGES`)—— 那时增量只覆盖了积压的一部分,拿它当「全部」 * 而服务端只给严格 `>`,那一毫秒的尾巴再也追不回来;客户端分不出干净尾巴与这个洞),或翻满了
* 就是把红点做小; * `MessageRepository.MAX_INCREMENT_PAGES`。那时增量只覆盖了积压的一部分,拿它当「全部」就是
* 把红点做小;
* 3. **请求失败**:含会话已在别处被删掉的 `404` —— 那一档要靠整表把这一行从列表里去掉。 * 3. **请求失败**:含会话已在别处被删掉的 `404` —— 那一档要靠整表把这一行从列表里去掉。
* *
* 三条都是「增量给不出可信答案」⇒ 去问唯一权威。**一次批次最多回落一次**:整批共用这一个失败 * 三条都是「增量给不出可信答案」⇒ 去问唯一权威。**一次批次最多回落一次**:整批共用这一个失败
@@ -60,34 +60,53 @@ class MessageIncrementRepositoryTest {
} }
@Test @Test
fun `a full page is followed by one more request and the empty page ends the chase`() = runBlocking { fun `an empty first page is a clean tail`() = runBlocking {
backend.enqueue(200, """{"data":[]}""")
val increment = repository().loadIncrement("web", after = 300).getOrThrow()
// 第一页就空 ⇒ 一条新消息都没有,没有可少给的:这是真的追到尾巴,游标原地不动。
assertEquals(1, backend.uris.size)
assertTrue(backend.uris.single(), backend.uris.single().contains("after=300"))
assertTrue(increment.complete)
assertEquals(300L, increment.cursor)
assertTrue(increment.messages.isEmpty())
}
@Test
fun `an empty page after a full page is not proof of a clean tail`() = runBlocking {
backend.enqueue(200, """{"data":[${message("m1", 100)}, ${message("m2", 200)}]}""") backend.enqueue(200, """{"data":[${message("m1", 100)}, ${message("m2", 200)}]}""")
backend.enqueue(200, """{"data":[]}""") backend.enqueue(200, """{"data":[]}""")
val increment = repository().loadIncrement("web", after = 0, pageSize = 2).getOrThrow() val increment = repository().loadIncrement("web", after = 0, pageSize = 2).getOrThrow()
// 满页 ⇒ 多问一次;那一次拿到空页,于是「追到了」而不是「翻不动了」。 // 满页 ⇒ 多问一次,第二次拿到空页。但客户端**分不出**这次空页是「干净的尾巴」还是
// 「空页底下还压着同一毫秒的几条」:服务端只给严格 `> cursor`,被页边界劈开的那一毫秒
// 再也追不回来。分不出来就只能说「没追完」,把「要不要回落整表」交回调用方。
assertEquals(2, backend.uris.size) assertEquals(2, backend.uris.size)
assertTrue(backend.uris[1], backend.uris[1].contains("after=200")) assertTrue(backend.uris[1], backend.uris[1].contains("after=200"))
assertTrue(increment.complete) assertFalse(increment.complete)
assertEquals(200L, increment.cursor) assertEquals(200L, increment.cursor)
assertEquals(listOf("m1", "m2"), increment.messages.map { it.id }) assertEquals(listOf("m1", "m2"), increment.messages.map { it.id })
} }
@Test @Test
fun `a full page stuck on one millisecond is reported as incomplete`() = runBlocking { fun `a full page stuck on one millisecond is reported as incomplete`() = runBlocking {
// 同毫秒并列且超过页宽:纯 `createdAt` 游标推不动,服务端只会把同一批行再给一次。 // 真后端的形状:m1 / m2 / m3 三条落在同一毫秒,页宽 2。第一页给 m1、m2,第二页问
val sameMillisecond = """{"data":[${message("m1", 500)}, ${message("m2", 500)}]}""" // `after=500` —— 严格 `>` 下后端只会回空页,m3@500 永远拿不到(这里不入队 m3:队列里
backend.enqueue(200, sameMillisecond) // 那份空响应就是真后端在 `after=500` 时会给的答案)。
backend.enqueue(200, sameMillisecond) backend.enqueue(200, """{"data":[${message("m1", 500)}, ${message("m2", 500)}]}""")
backend.enqueue(200, """{"data":[]}""")
val increment = repository().loadIncrement("web", after = 100, pageSize = 2).getOrThrow() val increment = repository().loadIncrement("web", after = 100, pageSize = 2).getOrThrow()
assertFalse(increment.complete) assertFalse(increment.complete)
// 游标没有越过那一毫秒(推不动就不装作推得动)。 // 游标停在那一毫秒(推不动就不装作推得动)。
assertEquals(500L, increment.cursor) assertEquals(500L, increment.cursor)
// 同一批行被问了两次,按 id 去重后仍是两条 —— 不能因为翻了两页就把条数翻倍。
assertEquals(listOf("m1", "m2"), increment.messages.map { it.id }) assertEquals(listOf("m1", "m2"), increment.messages.map { it.id })
// 线上证据:第二次真的带上了 `after=500`,即「m3 拿不到」的那一问。
assertEquals(2, backend.uris.size)
assertTrue(backend.uris[1], backend.uris[1].contains("after=500"))
} }
@Test @Test
@@ -130,8 +130,9 @@ class SessionIncrementTest {
@Test @Test
fun `messages at or before the read cursor are already read on the server`() { fun `messages at or before the read cursor are already read on the server`() {
// 服务端口径是 `count(created_at > read_at)`,而 `read_at` 是账号级单一游标: // 与服务端**同源的只是过滤口径**(`count(created_at > read_at)`),不是基数:
// 这台机器点开过它(游标推到「现在」)之后别的设备又发了消息,那段区间服务端算已读。 // `read_at` 是账号级单一游标,这台机器点开过它(游标推到「现在」)之后别的设备又发了消息,
// 那段区间服务端算已读 —— 加进角标就多出一个红点,所以先按 readAt 筛。
val sessions = listOf(item(id = "web", lastMessageAt = 100)) val sessions = listOf(item(id = "web", lastMessageAt = 100))
val updated = sessions.applyIncrement( val updated = sessions.applyIncrement(
@@ -145,6 +146,24 @@ class SessionIncrementTest {
assertEquals(1, updated.single().unreadCount) assertEquals(1, updated.single().unreadCount)
} }
@Test
fun `a read elsewhere while offline over-counts until the next whole table`() {
// 已接受的边界(见 `applyIncrement` 的 KDoc),钉在这里是为了它不会被当成「已修好」:
// 底数 `unreadCount` 来自上一次整表刷新,而 `read_at` 是账号级的。断线期间他端把这个会话
// 读掉了,那条 `session_read` 帧本地**永久收不到**(断线不补帧)⇒ 本端 readAt 还是 null,
// 整批都进了角标(4 + 3 = 7),而服务端此刻算 0。漂移到下一次整表刷新自愈,这里不改行为。
val sessions = listOf(item(id = "web", lastMessageAt = 100, unreadCount = 4))
val updated = sessions.applyIncrement(
"web",
from = 100,
readAt = null,
increment = increment(listOf(message("m1", 200), message("m2", 300), message("m3", 400))),
)
assertEquals(7, updated.single().unreadCount)
}
@Test @Test
fun `an unknown read cursor counts everything in the increment`() { fun `an unknown read cursor counts everything in the increment`() {
val sessions = listOf(item(id = "web", lastMessageAt = 100)) val sessions = listOf(item(id = "web", lastMessageAt = 100))