feat(W45): 发送阶梯改成直接试 REST，删掉 W41 的连通性探测
为什么换成直接尝试 - W41 的阶梯是「WS 未就绪 ⇒ 先探 GET /api/v1/health ⇒ 通才走 REST」。 - 真机回归（devtools/ws_block_proxy.py 只掐 /ws/chat，HTTP 全放行）发现消息进了队列而不是回落 REST；而 ProbeClientTest 证明那条链在 JVM 上是通的，Unreachable 生在连接之前，三种成因 （建 client 抛 / 连上但一个字没写 / ConnectException）在外同形，那一轮无法定论。 - 结论：不该有探测。探测只是「能不能连上」的替身，而真正的 REST 请求本身就是最直接的检验 ——试成了消息还顺手发出去了；试败了进队列（内容一个字节不丢，重发仍是同一个 clientMsgId）。 探测唯一省下的是「完全无网时少发一个注定失败的请求」，代价却是一个独立 client、一套独立超时 预算和一整套新的失败面。那个探针 client 还不复用连接池，每次都要现做一次冷握手，预算比主链路 紧 4 倍（2.5 秒整通 vs 10 秒连接）——今天这个 bug 的形态就长在这上面。 改了什么 - 路由：wsDispatched ⇒ Ws（成功路径逐字不变）；Offline ⇒ Queue；其余三档未就绪 ⇒ 直接试 REST。 - 直接试 REST 走既有的 REST 发送路径与请求体（同一份 chatRequest / 同一个 clientMsgId）：服务端 收下（200 stream:true / 202 queued / 200 duplicate）⇒ 完成、不进队列；网络类失败与没等到裁决 ⇒ 进队列；服务端给过裁决（409 / 500 / 503 / 400 与 401）⇒ 回滚 + 既有文案，口径不变。 - 判据纯函数化：sendRouteFor 去掉探测入参；新增 restAttemptLanding（Result<ChatAck>? ⇒ 四档落点）； ChatAck.isAccepted 从 ChatViewModel 的私有方法搬到判据文件。 - 删掉探测那一套：ConnectivityRepository / HealthApi / HealthDto / ApiClientFactory.createProbe / PROBE_TIMEOUT_MILLIS / Reachability / reachabilityOf / ChatViewModel.probeReachability / AppContainer 的接线（删前逐个确认过没有别的调用方）。 - 超时：一个 client、两个超时。client 上的连接超时 10 秒改成 2.5 秒；回落那一次投递的整通上限仍然 是 10 秒（REST_FALLBACK_TIMEOUT_MILLIS 一个数都没动）。 - 测试：SendFallbackTest 按新事实重写；ProbeClientTest 删掉，其中仍然成立的事实搬到新的 RestAttemptClientTest（真 OkHttp + 真 Retrofit + JDK HttpServer）。 为什么是这两个超时值 - 连接超时 2.5 秒：正常网络下一次 TCP 握手是毫秒到几十毫秒量级，2.5 秒是「人开始觉得卡」之前的 量级，也与 W41 那 2.5 秒的探针预算对齐 ⇒ 换成直接试 REST 之后「无网时用户最多等多久」没有变长 （完全无网时 ≤ ~2.5 秒进队列）。绝不许回到 10 秒：那会让每一条离线消息的用户都白等十倍的时间。 - 整通上限 10 秒：它管的是另一档——「连上了但一直不回话」（半死的代理、接住连接却不回应的 服务端），到点也按网络类失败进队列。它同时是用户干等时间的上界，不许拉长。 - 整通上限放在调用点而不是 client 的 callTimeout：这个 client 是共用的，长文本 / 内联图片那两条 本来就该走 REST 的路用的是同一个，client 级 callTimeout 会把它们一并框住，等于给一条本轮不打算 碰的路引入新失败面。 为什么 Offline 仍然不试（保留 W41 的决定） - 该档的语义是「握手 401 且续期也失败」（终态、不自动重连），会话已知是死的：试 REST 只会拿到 401 并落到「回滚 + 一句误导人的鉴权错误」，把「已排队不会丢」换成更差的结果。W41 的理由 （不鉴权的健康端点必然回 200 ⇒ 假通）随探测一起消失了，但结论换到直接试这边一字不用改。 验证 - gradlew.bat assembleDebug：BUILD SUCCESSFUL - gradlew.bat testDebugUnitTest：1153 个用例，failures=0，errors=0 （SendFallbackTest 14 例、RestAttemptClientTest 6 例全绿） 未做 - 没有 push。devdocs/05 的 W41 / W44 段落仍写着探测（那份由主会话维护，本轮只在回复里点出过时 的句子，没有改它）。
This commit is contained in:
AskaEth committed
1 parent
2bee0b52b0
commit
ec8d54f3a5
12 files changed
+720
-798
No files matched your search
@@ -1,200 +0,0 @@
|
||||
package top.yeij.cyrene.data.api
|
||||
|
||||
import com.sun.net.httpserver.HttpExchange
|
||||
import com.sun.net.httpserver.HttpServer
|
||||
import kotlinx.coroutines.runBlocking
|
||||
import org.junit.After
|
||||
import org.junit.Assert.assertEquals
|
||||
import org.junit.Assert.assertNull
|
||||
import org.junit.Assert.assertTrue
|
||||
import org.junit.Test
|
||||
import top.yeij.cyrene.core.error.CyreneError
|
||||
import top.yeij.cyrene.data.repository.ConnectivityRepository
|
||||
import top.yeij.cyrene.ui.chat.Reachability
|
||||
import top.yeij.cyrene.ui.chat.reachabilityOf
|
||||
import java.net.InetSocketAddress
|
||||
import java.net.ServerSocket
|
||||
import java.util.concurrent.CopyOnWriteArrayList
|
||||
import java.util.concurrent.Executors
|
||||
|
||||
/**
|
||||
* W41 连通性探测那条链的**运行时接线**：`ApiClientFactory.createProbe` → [HealthApi] →
|
||||
* `ConnectivityRepository.probe` → `reachabilityOf`。
|
||||
*
|
||||
* 端到端跑**真 OkHttp + 真 Retrofit + 真 HTTP 服务端**（JDK 自带的 `com.sun.net.httpserver`，
|
||||
* 与 [top.yeij.cyrene.data.repository.ChatRepositoryTest] 同一套做法，不引任何新依赖）。
|
||||
*
|
||||
* ## 它为什么存在
|
||||
* W41 的真机回归里，`GET /api/v1/health` 在代理日志里**一条都没有**：WS 断（`Reconnecting`）、
|
||||
* HTTP 通（sessions / messages / enums / personas 全 200），这条阶梯却选了队列而不是回落 REST。
|
||||
* 那一轮只验过**纯判据**（`sendRouteFor` / `reachabilityOf` 是纯函数，全绿），
|
||||
* 「这条链真能发出请求吗」**一次都没被跑过**——README 里那句「纯判据全绿、接线却断了」说的就是这件事。
|
||||
* 本文件补上这一段：**探测必须真的在 socket 上留下一个 `GET /api/v1/health`**，而不只是判据算对。
|
||||
*
|
||||
* ## 它钉住的四条事实（每一条都是判据，不是描述）
|
||||
* 1. 请求**真的发出去**，且落到 `GET <baseUrl>api/v1/health`（baseUrl 是建 client 时按当前服务器
|
||||
* 地址现给的，不是空）。
|
||||
* 2. 探测**不带令牌**（`Authorization` 缺席；该端点不鉴权，带一个过期令牌去问会把「后端好好的」
|
||||
* 读成不通）——这是 [ApiClientFactory.createProbe] 与通用 [ApiClientFactory.create] 的**唯一**
|
||||
* 语义差别，必须逐字钉住。
|
||||
* 3. `200 + {"data":{...}}` ⇒ `probe()` 回 `Result.success(true)` ⇒ [Reachability.Reachable]；
|
||||
* 非 2xx 与「连不上」⇒ `Result.failure` ⇒ [Reachability.Unreachable]（fail-closed）。
|
||||
* 4. 服务端**接了连接却不回话**时，探测在 [PROBE_TIMEOUT_MILLIS] 量级返回失败——它是一道
|
||||
* **整通上限**（`callTimeout`），绝不会把用户的发送挂在一次等不到头的探测上。
|
||||
*/
|
||||
class ProbeClientTest {
|
||||
|
||||
private val backend = FakeHealthBackend()
|
||||
|
||||
@After
|
||||
fun tearDown() {
|
||||
backend.close()
|
||||
}
|
||||
|
||||
private fun healthApi(baseUrl: String = backend.baseUrl) =
|
||||
ApiClientFactory.createProbe(baseUrl, HealthApi::class.java)
|
||||
|
||||
@Test
|
||||
fun `the probe really puts a GET api v1 health on the wire`() = runBlocking {
|
||||
backend.reply(200, """{"data":{"status":"ok"}}""")
|
||||
|
||||
healthApi().health()
|
||||
|
||||
// 请求行不含 HTTP 版本（HttpExchange 只给 method + URI），判据是「方法 + 路径」。
|
||||
val line = backend.requestLines.single()
|
||||
assertEquals("GET /api/v1/health", line)
|
||||
}
|
||||
|
||||
@Test
|
||||
fun `the probe carries no token because the endpoint is public`() = runBlocking {
|
||||
backend.reply(200, """{"data":{"status":"ok"}}""")
|
||||
|
||||
healthApi().health()
|
||||
|
||||
assertNull(backend.headers.single()["Authorization"])
|
||||
}
|
||||
|
||||
@Test
|
||||
fun `a 200 with data is reachable`() = runBlocking {
|
||||
backend.reply(200, """{"data":{"status":"ok"}}""")
|
||||
|
||||
val probed = ConnectivityRepository(healthApi()).probe()
|
||||
|
||||
assertEquals(Result.success(true), probed)
|
||||
assertEquals(Reachability.Reachable, reachabilityOf(probed))
|
||||
}
|
||||
|
||||
@Test
|
||||
fun `a non 2xx is unreachable`() = runBlocking {
|
||||
backend.reply(503, """{"error":{"code":"unavailable","message":"down"}}""")
|
||||
|
||||
val probed = ConnectivityRepository(healthApi()).probe()
|
||||
|
||||
assertTrue("非 2xx 该是失败，实际 $probed", probed.isFailure)
|
||||
assertEquals(Reachability.Unreachable, reachabilityOf(probed))
|
||||
}
|
||||
|
||||
@Test
|
||||
fun `a 2xx without data is unreachable rather than a false reachable`() = runBlocking {
|
||||
// 回执解析不出来 = 「后端没回我一句我能读懂的话」⇒ fail-closed 到不通。
|
||||
backend.reply(200, "{}")
|
||||
|
||||
val probed = ConnectivityRepository(healthApi()).probe()
|
||||
|
||||
assertEquals(
|
||||
CyreneError.Local(CyreneError.LocalKind.Parse),
|
||||
(probed.exceptionOrNull() as CyreneException).error,
|
||||
)
|
||||
assertEquals(Reachability.Unreachable, reachabilityOf(probed))
|
||||
}
|
||||
|
||||
@Test
|
||||
fun `a server that accepts but never answers trips the call timeout`() = runBlocking {
|
||||
// 接住连接、一个字都不回：模拟「半死的代理 / 黑洞服务端」。
|
||||
// 判据有两头——**必须失败**（fail-closed），且**在 callTimeout 量级返回**
|
||||
// （它是整通上限；绝不能退化到共用客户端那条 60s 读超时上，那会把用户的发送挂住）。
|
||||
BlackHole().use { hole ->
|
||||
val started = System.currentTimeMillis()
|
||||
|
||||
val probed = ConnectivityRepository(
|
||||
ApiClientFactory.createProbe(hole.baseUrl, HealthApi::class.java),
|
||||
).probe()
|
||||
|
||||
val elapsed = System.currentTimeMillis() - started
|
||||
assertTrue("该失败，实际 $probed", probed.isFailure)
|
||||
assertEquals(Reachability.Unreachable, reachabilityOf(probed))
|
||||
assertTrue(
|
||||
"该在探测上限附近返回，实际 ${elapsed}ms",
|
||||
elapsed >= PROBE_TIMEOUT_MILLIS - 500,
|
||||
)
|
||||
assertTrue(
|
||||
"绝不该等到共用客户端那条 60s 读超时，实际 ${elapsed}ms",
|
||||
elapsed < 15_000,
|
||||
)
|
||||
}
|
||||
}
|
||||
|
||||
/** 照着 `reply(...)` 回话的假健康端点：记下**请求行原文**与请求头，绝不校验鉴权。 */
|
||||
private inner class FakeHealthBackend : AutoCloseable {
|
||||
|
||||
private val server = HttpServer.create(InetSocketAddress("127.0.0.1", 0), 0).apply {
|
||||
executor = Executors.newFixedThreadPool(2)
|
||||
createContext("/api/v1/health", ::handle)
|
||||
start()
|
||||
}
|
||||
|
||||
val baseUrl: String get() = "http://127.0.0.1:${server.address.port}/"
|
||||
|
||||
val requestLines = CopyOnWriteArrayList<String>()
|
||||
val headers = CopyOnWriteArrayList<Map<String, String>>()
|
||||
|
||||
@Volatile
|
||||
private var status = 200
|
||||
|
||||
@Volatile
|
||||
private var body = "{}"
|
||||
|
||||
fun reply(status: Int, body: String) {
|
||||
this.status = status
|
||||
this.body = body
|
||||
}
|
||||
|
||||
private fun handle(exchange: HttpExchange) {
|
||||
requestLines += "${exchange.requestMethod} ${exchange.requestURI}"
|
||||
headers += exchange.requestHeaders.entries.associate { it.key to it.value.joinToString(", ") }
|
||||
val payload = body.toByteArray(Charsets.UTF_8)
|
||||
exchange.responseHeaders.add("Content-Type", "application/json")
|
||||
exchange.sendResponseHeaders(status, payload.size.toLong())
|
||||
exchange.responseBody.use { it.write(payload) }
|
||||
}
|
||||
|
||||
override fun close() {
|
||||
server.stop(0)
|
||||
}
|
||||
}
|
||||
|
||||
/** 只接连接、不回话的服务端（`callTimeout` 的判据用它）。 */
|
||||
private class BlackHole : AutoCloseable {
|
||||
|
||||
private val server = ServerSocket(0)
|
||||
private val held = CopyOnWriteArrayList<java.net.Socket>()
|
||||
|
||||
/** 收连接的后台线程：`accept` 之后什么都不做，把连接攥住。 */
|
||||
private val accepter = Thread {
|
||||
runCatching {
|
||||
while (!server.isClosed) held += server.accept()
|
||||
}
|
||||
}.apply {
|
||||
isDaemon = true
|
||||
start()
|
||||
}
|
||||
|
||||
val baseUrl: String get() = "http://127.0.0.1:${server.localPort}/"
|
||||
|
||||
override fun close() {
|
||||
runCatching { server.close() }
|
||||
accepter.interrupt()
|
||||
held.forEach { runCatching { it.close() } }
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,210 @@
|
||||
package top.yeij.cyrene.data.api
|
||||
|
||||
import com.sun.net.httpserver.HttpExchange
|
||||
import com.sun.net.httpserver.HttpServer
|
||||
import kotlinx.coroutines.runBlocking
|
||||
import kotlinx.coroutines.withTimeoutOrNull
|
||||
import org.junit.After
|
||||
import org.junit.Assert.assertEquals
|
||||
import org.junit.Assert.assertTrue
|
||||
import org.junit.Test
|
||||
import top.yeij.cyrene.data.api.dto.ChatRequest
|
||||
import top.yeij.cyrene.data.repository.ChatAck
|
||||
import top.yeij.cyrene.data.repository.ChatRepository
|
||||
import top.yeij.cyrene.ui.chat.REST_FALLBACK_TIMEOUT_MILLIS
|
||||
import top.yeij.cyrene.ui.chat.RestAttemptLanding
|
||||
import top.yeij.cyrene.ui.chat.restAttemptLanding
|
||||
import java.net.InetSocketAddress
|
||||
import java.net.ServerSocket
|
||||
import java.util.concurrent.CopyOnWriteArrayList
|
||||
import java.util.concurrent.Executors
|
||||
|
||||
/**
|
||||
* 「WS 发不出去就直接试 REST」这条路的**运行时接线**：`ApiClientFactory.create`（主 REST client）→
|
||||
* [ChatApi] → [ChatRepository.send] → [restAttemptLanding]。
|
||||
*
|
||||
* 端到端跑**真 OkHttp + 真 Retrofit + 真 HTTP 服务端**（JDK 自带的 `com.sun.net.httpserver`，与
|
||||
* [top.yeij.cyrene.data.repository.ChatRepositoryTest] 同一套做法，不引任何新依赖）。
|
||||
*
|
||||
* ## 它为什么存在（W45 接下 W41 的位置）
|
||||
* W41 那道探测的运行时接线由 `ProbeClientTest` 钉住，本轮连探测一起删了。但那个文件里有一条教训
|
||||
* 是**和方法无关**的：真机回归里「请求到底有没有写出去」曾经无人可证（代理日志一条都没有，而
|
||||
* 「压根没发」与「发出去之前就死了」在外面完全同形）。所以新判据同样要有一条**在真 socket 上**
|
||||
* 的证据，只是对象换成了真正的那次 `POST /api/v1/chat`：
|
||||
*
|
||||
* ## 它钉住的五条事实（每一条都是判据，不是描述）
|
||||
* 1. 这次尝试**真的写出去**了，落到 `POST <baseUrl>api/v1/chat`（用**主 REST client**，不是另
|
||||
* 建一个探针 client）。
|
||||
* 2. 请求体里带的 `clientMsgId` **就是那条乐观气泡的 `localId`**：这是「试败进队列后重发也只落
|
||||
* 一条」的全部依据（服务端按 `(session_id, client_msg_id)` 仲裁），也是「WS 那帧没交出去、
|
||||
* REST 这条带上同一个键」这件事唯一的可验证形态。
|
||||
* 3. 它**带着令牌**（`Authorization: Bearer …`）——与 W41 那个不鉴权的探针**恰好相反**：这条走的
|
||||
* 是真正的业务端点（带 `X-API-Version` 与 401 续期），不是一次健康检查。
|
||||
* 4. 服务端收下（`200 {stream:true}` / `202 {queued:true}` / `200 {duplicate:true}`）⇒
|
||||
* [RestAttemptLanding.Accepted]（**不进队列**）；服务端拒绝（`409 chat_busy` 等）⇒
|
||||
* [RestAttemptLanding.Refused]（回滚）。
|
||||
* 5. 连不上（对端不在听）⇒ 失败，且在**连接超时量级**内返回，落到 [RestAttemptLanding.Unsent]
|
||||
* （进队列）——绝不会把用户的发送挂在一整通上限上。
|
||||
*/
|
||||
class RestAttemptClientTest {
|
||||
|
||||
private val backend = FakeChatBackend()
|
||||
|
||||
@After
|
||||
fun tearDown() {
|
||||
backend.close()
|
||||
}
|
||||
|
||||
/** 与生产同形：主 REST client（带令牌与 401 策略），只是 baseUrl 指向本地假端点。 */
|
||||
private fun repository(baseUrl: String = backend.baseUrl) = ChatRepository(
|
||||
ApiClientFactory.create(
|
||||
baseUrl = baseUrl,
|
||||
service = ChatApi::class.java,
|
||||
tokenProvider = { TOKEN },
|
||||
),
|
||||
)
|
||||
|
||||
/** 与生产同形的一次尝试：带整通上限的、原样交给 [restAttemptLanding] 的那个 `Result`。 */
|
||||
private suspend fun attempt(
|
||||
repository: ChatRepository,
|
||||
clientMsgId: String = CLIENT_MSG_ID,
|
||||
): Result<ChatAck>? = withTimeoutOrNull(REST_FALLBACK_TIMEOUT_MILLIS) {
|
||||
repository.send(
|
||||
ChatRequest(
|
||||
sessionId = SESSION_ID,
|
||||
text = "你好",
|
||||
deviceId = "voiceassist:cyrene:v1",
|
||||
stream = true,
|
||||
clientMsgId = clientMsgId,
|
||||
),
|
||||
)
|
||||
}
|
||||
|
||||
@Test
|
||||
fun `the attempt really puts a POST api v1 chat on the wire`() = runBlocking {
|
||||
backend.reply(200, """{"data":{"stream":true}}""")
|
||||
|
||||
attempt(repository())
|
||||
|
||||
// 请求行不含 HTTP 版本（HttpExchange 只给 method + URI），判据是「方法 + 路径」。
|
||||
assertEquals("POST /api/v1/chat", backend.requestLines.single())
|
||||
}
|
||||
|
||||
@Test
|
||||
fun `the attempt carries the very same client message id as the ws frame had`() = runBlocking {
|
||||
backend.reply(200, """{"data":{"stream":true}}""")
|
||||
|
||||
attempt(repository(), clientMsgId = "local-42")
|
||||
|
||||
val sent = ApiClientFactory.json.decodeFromString(ChatRequest.serializer(), backend.bodies.single())
|
||||
// 「发过两次也只落一条」的全部依据：REST 这条**必须**带同一条乐观气泡的键。
|
||||
assertEquals("local-42", sent.clientMsgId)
|
||||
assertEquals(SESSION_ID, sent.sessionId)
|
||||
// `stream:true` 也是必须的：`stream:false` 是同步的，服务端没有写超时，客户端读超时会先超时。
|
||||
assertEquals(true, sent.stream)
|
||||
}
|
||||
|
||||
@Test
|
||||
fun `the attempt carries the token because this is the real chat endpoint`() = runBlocking {
|
||||
backend.reply(200, """{"data":{"stream":true}}""")
|
||||
|
||||
attempt(repository())
|
||||
|
||||
val headers = backend.headers.single()
|
||||
assertEquals("Bearer $TOKEN", headers["authorization"])
|
||||
assertEquals("1", headers["x-api-version"])
|
||||
}
|
||||
|
||||
@Test
|
||||
fun `acks that mean the server took the message land as done and never queue`() = runBlocking {
|
||||
// 三种「服务端已经收下」的回执形状（契约 v0.56 的响应面），逐个走一遍真 HTTP。
|
||||
val accepted = listOf(
|
||||
200 to """{"data":{"stream":true}}""",
|
||||
202 to """{"data":{"stream":true,"queued":true,"queueDepth":1}}""",
|
||||
200 to """{"data":{"duplicate":true,"messageId":"m-9","sessionId":"web"}}""",
|
||||
)
|
||||
accepted.forEach { (status, body) ->
|
||||
backend.reply(status, body)
|
||||
|
||||
val landing = restAttemptLanding(attempt(repository()))
|
||||
|
||||
assertTrue("$status $body 该是「服务端收下了」，实际 $landing", landing is RestAttemptLanding.Accepted)
|
||||
}
|
||||
}
|
||||
|
||||
@Test
|
||||
fun `a refusal from the server lands as a rollback instead of a hidden retry`() = runBlocking {
|
||||
backend.reply(409, """{"error":{"code":"chat_busy","message":"会话正忙，待处理消息已满"}}""")
|
||||
|
||||
val landing = restAttemptLanding(attempt(repository()))
|
||||
|
||||
// 服务端**回了话**：这条没被受理、没落库 ⇒ 回滚 + 文案，既不进队列也不偷偷重发。
|
||||
assertEquals(RestAttemptLanding.Refused(ChatAck.QueueFull), landing)
|
||||
}
|
||||
|
||||
@Test
|
||||
fun `an endpoint that is not listening fails fast and lands in the queue`() = runBlocking {
|
||||
// 一个没人听的端口（拿一个空闲端口号之后立刻放掉）：连不上 ⇒ 失败。
|
||||
// 判据两头——**必须落到队列**（内容一个字节不丢），且**在短连接超时量级内返回**
|
||||
// （绝不能退化成整通上限，那会把用户的发送挂住）。
|
||||
val deadPort = ServerSocket(0).use { it.localPort }
|
||||
val started = System.currentTimeMillis()
|
||||
|
||||
val landing = restAttemptLanding(attempt(repository("http://127.0.0.1:$deadPort/")))
|
||||
|
||||
val elapsed = System.currentTimeMillis() - started
|
||||
assertEquals(RestAttemptLanding.Unsent, landing)
|
||||
assertTrue(
|
||||
"连不上该很快返回（连接超时量级），实际 ${elapsed}ms",
|
||||
elapsed < CONNECT_TIMEOUT_MILLIS + 3_000L,
|
||||
)
|
||||
}
|
||||
|
||||
/** 照着 `reply(...)` 回话的假 chat 端点：记下**请求行原文**、请求头原文与请求体原文。 */
|
||||
private inner class FakeChatBackend : AutoCloseable {
|
||||
|
||||
private val server = HttpServer.create(InetSocketAddress("127.0.0.1", 0), 0).apply {
|
||||
executor = Executors.newFixedThreadPool(2)
|
||||
createContext("/api/v1/chat", ::handle)
|
||||
start()
|
||||
}
|
||||
|
||||
val baseUrl: String get() = "http://127.0.0.1:${server.address.port}/"
|
||||
|
||||
val requestLines = CopyOnWriteArrayList<String>()
|
||||
/** 请求头按**小写键**记录：JDK 的 `Headers` 会把键规范化成 `X-Api-Version` 这种形状。 */
|
||||
val headers = CopyOnWriteArrayList<Map<String, String>>()
|
||||
val bodies = CopyOnWriteArrayList<String>()
|
||||
|
||||
@Volatile
|
||||
private var status = 200
|
||||
|
||||
@Volatile
|
||||
private var body = "{}"
|
||||
|
||||
fun reply(status: Int, body: String) {
|
||||
this.status = status
|
||||
this.body = body
|
||||
}
|
||||
|
||||
private fun handle(exchange: HttpExchange) {
|
||||
requestLines += "${exchange.requestMethod} ${exchange.requestURI}"
|
||||
headers += exchange.requestHeaders.entries.associate { it.key.lowercase() to it.value.joinToString(", ") }
|
||||
bodies += String(exchange.requestBody.readBytes(), Charsets.UTF_8)
|
||||
val payload = body.toByteArray(Charsets.UTF_8)
|
||||
exchange.responseHeaders.add("Content-Type", "application/json")
|
||||
exchange.sendResponseHeaders(status, payload.size.toLong())
|
||||
exchange.responseBody.use { it.write(payload) }
|
||||
}
|
||||
|
||||
override fun close() {
|
||||
server.stop(0)
|
||||
}
|
||||
}
|
||||
|
||||
private companion object {
|
||||
const val SESSION_ID = "web"
|
||||
const val CLIENT_MSG_ID = "local-1"
|
||||
const val TOKEN = "test-token"
|
||||
}
|
||||
}
|
||||
@@ -1,193 +1,203 @@
|
||||
package top.yeij.cyrene.ui.chat
|
||||
|
||||
import java.net.ConnectException
|
||||
import java.net.SocketTimeoutException
|
||||
import java.net.UnknownHostException
|
||||
import org.junit.Assert.assertEquals
|
||||
import org.junit.Assert.assertNotEquals
|
||||
import org.junit.Assert.assertTrue
|
||||
import org.junit.Test
|
||||
import top.yeij.cyrene.core.error.CyreneError
|
||||
import top.yeij.cyrene.data.api.PROBE_TIMEOUT_MILLIS
|
||||
import top.yeij.cyrene.data.api.CONNECT_TIMEOUT_MILLIS
|
||||
import top.yeij.cyrene.data.api.CyreneException
|
||||
import top.yeij.cyrene.data.repository.ChatAck
|
||||
import top.yeij.cyrene.data.ws.WsState
|
||||
import top.yeij.cyrene.domain.chat.ChatSendRouting
|
||||
import top.yeij.cyrene.domain.chat.DraftDocument
|
||||
import top.yeij.cyrene.domain.chat.SendChannel
|
||||
|
||||
/**
|
||||
* 「未连接时先探测、能发就发，不通才排队」这条三级阶梯的判据
|
||||
* （[sendRouteFor] / [reachabilityOf] / [restFailureLanding]）。
|
||||
* 「WS 发不出去就**直接试 REST**」这套判据
|
||||
* （[sendRouteFor] / [restAttemptLanding] / [restFailureLanding] / [ChatAck.isAccepted]）。
|
||||
*
|
||||
* ## 背景：主人 2026-10-05 的第二次定案
|
||||
* W40 定的是「未连接就进队列」（[SendQueueTest] 那次），它解决的是**内容不被退回给用户**。
|
||||
* 这一轮的追问是「**未连接状态你就不能拿 httpAPI 尝试吗？**」，并明确为「可以走 health 端点验证
|
||||
* 连通性」。理由是网关只有一条：`/ws` 与 `POST /api/v1/chat` 是同一个 `:8080` 上的两条车道，
|
||||
* 握手被挡（代理不转发 Upgrade、中间设备掐长连接）时 REST 往往是通的——直接排队等于把
|
||||
* 「WS 这条车道暂时不通」当成「后端不可用」，让用户白等一次重连。
|
||||
* ## 背景：W45 把探测换成了直接尝试（主人的定案）
|
||||
* W41 的阶梯是「WS 未就绪 ⇒ 先探 `GET /api/v1/health` ⇒ 通才回落 REST」。真机回归
|
||||
* （`devtools/ws_block_proxy.py` 只掐 `/ws/chat`、HTTP 全放行）发现消息进了队列而不是回落 REST，
|
||||
* 而 `ProbeClientTest` 证明那条链在 JVM 上是通的、`Unreachable` 生在**连接之前**，三种成因在外
|
||||
* 同形 ⇒ 无法定论。判据与论证见 [sendRouteFor]。
|
||||
*
|
||||
* ## 本测试钉住五件事
|
||||
* 1. **优先级**（[sendRouteFor]）：帧交出去了 ⇒ 走 WS（连接那一格不再参与判断）；未就绪 + 探测通
|
||||
* ⇒ 走 REST 回落；未就绪 + 探测不通 / 超时 / 抛异常 / 没探过 ⇒ 进队列。
|
||||
* 2. **fail-closed**（[reachabilityOf]）：只有「探到 `true`」才算通，其余一切（含超时与异常）
|
||||
* 都归「不通」——探测这一层永远不因为拿不准而放行一次真发送。
|
||||
* 3. **探测必须短**：[PROBE_TIMEOUT_MILLIS] 落在 2~3 秒里（它直接加在用户的发送动作上）。
|
||||
* 4. **Offline 不探测**：那一档是「会话已知是死的」，健康端点不鉴权必然回 200，探了只会把
|
||||
* W40 的「已排队不会丢」换成一次回滚 + 一句误导人的鉴权错误。
|
||||
* 5. **回落失败后的落点**（[restFailureLanding]）：网络类失败（含超时）⇒ 进队列；服务端给过
|
||||
* 裁决的（拒绝 / 401）与配置类错误 ⇒ 今天那条回滚 + 文案。
|
||||
* 现在的判据里**没有任何探测输入**：路由只看「帧交出去了没有 + 这一刻的连接状态」两个事实，
|
||||
* REST 那一次的产物由 [restAttemptLanding] 判落点。
|
||||
*
|
||||
* ## 本测试钉住六件事
|
||||
* 1. **优先级**（[sendRouteFor]）：帧交出去了 ⇒ 走 WS（连接状态不再参与）；未就绪且不是
|
||||
* [WsState.Offline] ⇒ 直接试 REST；[WsState.Offline] ⇒ 进队列、**一个请求都不发**。
|
||||
* 2. **收下了就不进队列**：受理 / 续接排队 / 幂等重复（[ChatAck.isAccepted] 那一族）⇒
|
||||
* [RestAttemptLanding.Accepted]（收下还排队会白补发一次，界面上那句「排队中」也成了假话）。
|
||||
* 3. **fail-closed**：没等到裁决（整通上限到点 / 异常映射不出来）与网络类失败 ⇒
|
||||
* [RestAttemptLanding.Unsent]——**绝不许**把没上路的说成发出去了。
|
||||
* 4. **服务端给过裁决 ⇒ 回滚**：拒绝的回执（`409` / `500` / `503` / `400`）与 401 / 配置类异常。
|
||||
* 5. **两个超时**：连接超时落在 2~3 秒（无网时的等待上界），整通上限另设且不许拉长。
|
||||
* 6. **这条阶梯到不了「REST 专属」的消息**（有图 / 有附件一律在进入阶梯之前就被判成 REST）。
|
||||
*/
|
||||
class SendFallbackTest {
|
||||
|
||||
// —— 一、优先级：这一刻该走哪条路 ——
|
||||
|
||||
@Test
|
||||
fun `a frame handed to the socket walks the ws path whatever else says`() {
|
||||
// `sendChat` 回 true 就是「帧交给 socket 了」：连接状态那一格与探测结论都不再参与判断
|
||||
// （`sendChat` 判的就是它自己的状态）。这一档是**成功路径**，W41 一个字都不许动它——
|
||||
// 所以这里连探测结论都不给它：`null` 也得是 Ws。
|
||||
fun `a frame handed to the socket walks the ws path whatever the connection says`() {
|
||||
// `sendChat` 回 true 就是「帧交给 socket 了」：连接状态那一格不再参与判断（`sendChat` 判的
|
||||
// 就是它自己的状态）。这一档是**成功路径**，W41 与 W45 一个字都没动它——所以这里把五档
|
||||
// 连接状态逐个钉一遍。
|
||||
WsState.entries.forEach { connection ->
|
||||
listOf(null, Reachability.Reachable, Reachability.Unreachable).forEach { reachability ->
|
||||
assertEquals(
|
||||
"$connection 上帧已经交出去了，不该改走别的路",
|
||||
SendRoute.Ws,
|
||||
sendRouteFor(wsDispatched = true, connection = connection, reachability = reachability),
|
||||
)
|
||||
}
|
||||
assertEquals(
|
||||
"$connection 上帧已经交出去了，不该改走别的路",
|
||||
SendRoute.Ws,
|
||||
sendRouteFor(wsDispatched = true, connection = connection),
|
||||
)
|
||||
}
|
||||
}
|
||||
|
||||
@Test
|
||||
fun `not connected but reachable falls back to rest`() {
|
||||
// WS 没把帧交出去，而探测说后端这一刻在 ⇒ 用既有的 POST /api/v1/chat 把这条发出去。
|
||||
// 三档「未就绪」一视同仁：Idle / Connecting / Reconnecting 都只是「还没连上」，
|
||||
// 与「后端不可达」根本不是一件事。
|
||||
fun `a not yet connected state tries rest directly instead of probing`() {
|
||||
// 三档「还没连上」一视同仁：Idle / Connecting / Reconnecting 都只是「还没连上」，与
|
||||
// 「后端不可达」根本不是一件事。W45 起这里**直接试**——真正的 `POST /api/v1/chat` 就是那次
|
||||
// 检验（试成了消息还顺手发出去了），不再先花一个请求问「后端在不在」。
|
||||
listOf(WsState.Idle, WsState.Connecting, WsState.Reconnecting).forEach { connection ->
|
||||
assertEquals(
|
||||
"$connection 上探测通了，该回落 REST",
|
||||
"$connection 上该直接试 REST",
|
||||
SendRoute.Rest,
|
||||
sendRouteFor(wsDispatched = false, connection = connection, reachability = Reachability.Reachable),
|
||||
sendRouteFor(wsDispatched = false, connection = connection),
|
||||
)
|
||||
}
|
||||
}
|
||||
|
||||
@Test
|
||||
fun `not connected and unreachable queues`() {
|
||||
listOf(WsState.Idle, WsState.Connecting, WsState.Reconnecting).forEach { connection ->
|
||||
assertEquals(
|
||||
"$connection 上探测不通，该进队列",
|
||||
SendRoute.Queue,
|
||||
sendRouteFor(wsDispatched = false, connection = connection, reachability = Reachability.Unreachable),
|
||||
)
|
||||
}
|
||||
}
|
||||
|
||||
@Test
|
||||
fun `no probe result at all means queue, never a hopeful rest send`() {
|
||||
// `null` = 还没探过（或判据被改坏没喂结论）。没有证据说能发，就不发：
|
||||
// 这一档必须与「探测不通」同归一处，否则「拿不准」会变成一次注定失败的真发送。
|
||||
listOf(WsState.Idle, WsState.Connecting, WsState.Reconnecting, WsState.Offline).forEach { connection ->
|
||||
assertEquals(
|
||||
"$connection 上还没有探测结论，只许进队列",
|
||||
SendRoute.Queue,
|
||||
sendRouteFor(wsDispatched = false, connection = connection, reachability = null),
|
||||
)
|
||||
}
|
||||
}
|
||||
|
||||
// —— 二、探测的结论怎么算：fail-closed ——
|
||||
|
||||
@Test
|
||||
fun `only a confirmed answer counts as reachable`() {
|
||||
assertEquals(Reachability.Reachable, reachabilityOf(Result.success(true)))
|
||||
// 后端回了话但非 2xx / 回执解析不出来：仓库层折成 `false` 或 `failure`，都不是通。
|
||||
assertEquals(Reachability.Unreachable, reachabilityOf(Result.success(false)))
|
||||
}
|
||||
|
||||
@Test
|
||||
fun `a timing out probe is not reachable`() {
|
||||
// 硬要求：探测超时一律按「不通」处理（fail-closed 到队列）——**绝不许**把用户的发送
|
||||
// 卡在一个等不到头的探测上，也绝不许因为「说不定通」就先发一次 REST。
|
||||
assertEquals(
|
||||
Reachability.Unreachable,
|
||||
reachabilityOf(Result.failure(SocketTimeoutException("call timeout"))),
|
||||
)
|
||||
}
|
||||
|
||||
@Test
|
||||
fun `a probe that throws is not reachable, whatever it threw`() {
|
||||
// 连不上 / DNS / 别的任何异常：同一档。探测这一层**永远不抛异常给调用方**——
|
||||
// 它要回答的是一个 yes/no，抛出去只会让「这条消息去哪」这件事没有答案。
|
||||
val thrown = listOf(
|
||||
ConnectException("connection refused"),
|
||||
UnknownHostException("cyrene.invalid"),
|
||||
IllegalStateException("boom"),
|
||||
)
|
||||
thrown.forEach { e ->
|
||||
assertEquals("$e 不该被当成通", Reachability.Unreachable, reachabilityOf(Result.failure(e)))
|
||||
}
|
||||
}
|
||||
|
||||
@Test
|
||||
fun `an unset result is not reachable either`() {
|
||||
// `Result` 里没有值（既没成功也没给异常）也是一种「没拿到答案」⇒ 不通。
|
||||
assertEquals(Reachability.Unreachable, reachabilityOf(Result.failure(RuntimeException())))
|
||||
}
|
||||
|
||||
@Test
|
||||
fun `the probe timeout is short enough to sit inside a send`() {
|
||||
// 硬要求：探测必须有 2~3 秒量级的短超时。它直接加在用户的发送动作上（气泡已经上屏、
|
||||
// 用户等着这条消息的结论），所以既不能长到像卡死，也不能短到正常网络下误判成不通。
|
||||
assertTrue(
|
||||
"探测超时 $PROBE_TIMEOUT_MILLIS ms 落在 2~3 秒之外",
|
||||
PROBE_TIMEOUT_MILLIS in 2_000L..3_000L,
|
||||
)
|
||||
// 回落 REST 那一次投递另有一道上限，且必须比探测宽（它是真请求，不是探针）：
|
||||
// 两道合起来才是「没网时用户最多等多久」的上界。
|
||||
assertTrue(
|
||||
"回落 REST 的上限该比探测宽裕",
|
||||
REST_FALLBACK_TIMEOUT_MILLIS > PROBE_TIMEOUT_MILLIS,
|
||||
)
|
||||
}
|
||||
|
||||
// —— 三、Offline 那一档：不探测，直接排队 ——
|
||||
|
||||
@Test
|
||||
fun `offline queues without probing, because the session is already known dead`() {
|
||||
fun `offline queues without trying rest because the session is already known dead`() {
|
||||
// Offline 的语义是「握手 401 且续期也失败」（终态、不自动重连）⇒ 会话已知是死的。
|
||||
// 健康端点**不鉴权**（`Cyrene_v2/internal/services/gateway/routes.go:51` 挂在公开路由里），
|
||||
// 探它必然回 200 ⇒ 判成 Reachable ⇒ 发一次注定 401 的 REST ⇒ 把 W40 那句
|
||||
// 「已排队不会丢」换成一次回滚 + 一句会误导人的鉴权错误。所以这一档**根本不探**。
|
||||
listOf(null, Reachability.Reachable, Reachability.Unreachable).forEach { reachability ->
|
||||
assertEquals(
|
||||
"Offline 上不该因为探测通就去发 REST",
|
||||
SendRoute.Queue,
|
||||
sendRouteFor(wsDispatched = false, connection = WsState.Offline, reachability = reachability),
|
||||
)
|
||||
}
|
||||
// 这一档去试 REST 只会拿到 401、落到「回滚 + 一句会误导人的鉴权错误」，把 W40 那句
|
||||
// 「已排队不会丢」换成更差的结果。所以这一档**一次请求都不发**。
|
||||
//
|
||||
// W41 也判这一档进队列（它当时的理由是「不鉴权的健康端点必然回 200 ⇒ 假通」）；探测没了
|
||||
// 之后这个结论一字不用改——带令牌的 chat 端点回 401 是**真不通**，两条路指向同一件事。
|
||||
assertEquals(SendRoute.Queue, sendRouteFor(wsDispatched = false, connection = WsState.Offline))
|
||||
}
|
||||
|
||||
@Test
|
||||
fun `offline is the only state that skips the probe`() {
|
||||
// 三档「还没连上」都该走到探测那一步（判据只认 Connected 一个值，也排除 Offline 一个值）：
|
||||
// 将来连接状态多一档时，这一条会逼着有人来做决定，而不是静默继承。
|
||||
fun `offline is the only not connected state that skips the rest attempt`() {
|
||||
// 判据只认 Connected 一个值，也只把 Offline 排除掉一个值：将来连接状态多一档时，这一条会
|
||||
// 逼着有人来做决定，而不是静默继承。
|
||||
WsState.entries
|
||||
.filter { it != WsState.Connected && it != WsState.Offline }
|
||||
.forEach { connection ->
|
||||
assertEquals(
|
||||
"$connection 上该允许回落 REST",
|
||||
"$connection 上该允许直接试 REST",
|
||||
SendRoute.Rest,
|
||||
sendRouteFor(wsDispatched = false, connection = connection, reachability = Reachability.Reachable),
|
||||
sendRouteFor(wsDispatched = false, connection = connection),
|
||||
)
|
||||
}
|
||||
}
|
||||
|
||||
// —— 四、回落 REST 失败之后去哪 ——
|
||||
@Test
|
||||
fun `the whole routing table has two inputs and no probe among them`() {
|
||||
// 判据的全部输入就是这两个事实（**没有探测结论**——那是 W45 删掉的东西，这条断言连同
|
||||
// [sendRouteFor] 的签名一起把这件事钉死：多喂一个探测结论就编译不过）。每一格都有明确
|
||||
// 结论，没有第三档可能的输入。
|
||||
val table = listOf(
|
||||
Triple(true, WsState.Idle, SendRoute.Ws),
|
||||
Triple(true, WsState.Connecting, SendRoute.Ws),
|
||||
Triple(true, WsState.Reconnecting, SendRoute.Ws),
|
||||
Triple(true, WsState.Connected, SendRoute.Ws),
|
||||
Triple(true, WsState.Offline, SendRoute.Ws),
|
||||
Triple(false, WsState.Idle, SendRoute.Rest),
|
||||
Triple(false, WsState.Connecting, SendRoute.Rest),
|
||||
Triple(false, WsState.Reconnecting, SendRoute.Rest),
|
||||
// 这一格从阶梯上**到不了**：帧没交出去而连接就绪时 [wsSendOutcome] 判的是
|
||||
// `Failed`（回滚），sendRouteFor 根本不会被问到。写在这里是为了这张表是**全的**
|
||||
// （函数对五档连接状态都有定义），不是因为它是可达的一格。
|
||||
Triple(false, WsState.Connected, SendRoute.Rest),
|
||||
Triple(false, WsState.Offline, SendRoute.Queue),
|
||||
)
|
||||
table.forEach { (dispatched, connection, expected) ->
|
||||
assertEquals(
|
||||
"dispatched=$dispatched / $connection 该走 $expected",
|
||||
expected,
|
||||
sendRouteFor(wsDispatched = dispatched, connection = connection),
|
||||
)
|
||||
}
|
||||
}
|
||||
|
||||
// —— 二、直接试 REST 的落点 ——
|
||||
|
||||
@Test
|
||||
fun `network failures on the fallback queue instead of rolling back`() {
|
||||
fun `a server that took the message means done and never a queue entry`() {
|
||||
// 三种回执都是「服务端已经收下这条」：`200 {stream:true}`（起回合）、
|
||||
// `202 {queued:true}`（忙时续接排队，已落库）、`200 {duplicate:true}`（幂等键早就收过）。
|
||||
// ⇒ 了结，**不进队列**：收下了还排队会白补发一次（幂等键兜得住，但那是浪费，界面上那句
|
||||
// 「排队中 · 还没连上」也成了假话）。
|
||||
val accepted = listOf<ChatAck>(
|
||||
ChatAck.Accepted,
|
||||
ChatAck.Queued(queueDepth = 2),
|
||||
ChatAck.Duplicate(messageId = "m-1", sessionId = "web"),
|
||||
)
|
||||
accepted.forEach { ack ->
|
||||
assertEquals(
|
||||
"$ack 是服务端收下了，该是 Done",
|
||||
RestAttemptLanding.Accepted(ack),
|
||||
restAttemptLanding(Result.success(ack)),
|
||||
)
|
||||
}
|
||||
}
|
||||
|
||||
@Test
|
||||
fun `a rejected ack rolls back instead of being retried behind the user's back`() {
|
||||
// 服务端**回了话**就是它的裁决：`409 chat_busy`（队列满）/ `500 chat_unavailable` /
|
||||
// `503 chat_dispatch_failed` / `400` 别的明确拒绝。盲目重发换不来不同结果，而且把
|
||||
// 「服务端拒绝」说成「还没连上」是假话 ⇒ 回滚 + 各自的文案（文案在 onAck 那六支里）。
|
||||
val refused = listOf<ChatAck>(
|
||||
ChatAck.QueueFull,
|
||||
ChatAck.Unavailable("chat_unavailable", "chat service is not configured"),
|
||||
ChatAck.Rejected("bad_request", "文本、image 或 documents 至少需要一项"),
|
||||
)
|
||||
refused.forEach { ack ->
|
||||
assertEquals(
|
||||
"$ack 是服务端的裁决，该回滚",
|
||||
RestAttemptLanding.Refused(ack),
|
||||
restAttemptLanding(Result.success(ack)),
|
||||
)
|
||||
}
|
||||
}
|
||||
|
||||
@Test
|
||||
fun `an unauthorized or a configuration error rolls back because waiting will not fix it`() {
|
||||
// 401 与配置 / 协议族走的是**异常**那一支（`ChatRepository` 把 401 折成 failure），同样回滚：
|
||||
// 它们要的是用户重新登录 / 改设置 / 换地址，等下去不会变好；回滚时那句 userMessage 正是他
|
||||
// 需要看到的东西（把「令牌失效」说成「还没连上」会让他干等）。
|
||||
val unauthorized = CyreneError.Unauthorized(code = "invalid_token", message = "")
|
||||
assertEquals(
|
||||
RestAttemptLanding.Errored(unauthorized),
|
||||
restAttemptLanding(Result.failure(CyreneException(unauthorized))),
|
||||
)
|
||||
|
||||
val config = listOf(
|
||||
CyreneError.LocalKind.Tls,
|
||||
CyreneError.LocalKind.CleartextBlocked,
|
||||
CyreneError.LocalKind.InvalidUrl,
|
||||
CyreneError.LocalKind.Parse,
|
||||
CyreneError.LocalKind.Unknown,
|
||||
)
|
||||
config.forEach { kind ->
|
||||
val error = CyreneError.Local(kind)
|
||||
assertEquals("$kind 等下去不会变好，该回滚", RestFailureLanding.Rollback, restFailureLanding(error))
|
||||
assertEquals(
|
||||
RestAttemptLanding.Errored(error),
|
||||
restAttemptLanding(Result.failure(CyreneException(error))),
|
||||
)
|
||||
}
|
||||
}
|
||||
|
||||
@Test
|
||||
fun `a network failure puts the message in the queue`() {
|
||||
// 这一族都是「请求根本没被服务端读到」：等一下多半会好，而内容一点都不该退给用户。
|
||||
// 重发用的是同一个 clientMsgId，服务端按 (session_id, client_msg_id) 仲裁：真重复到达
|
||||
// 只回 chat_duplicate / {duplicate:true}，不会再落库、不会再起回合（契约 v0.56 §3.13）。
|
||||
// 重发用的是**同一个** `clientMsgId`，服务端按 (session_id, client_msg_id) 仲裁：真重复
|
||||
// 到达只回 chat_duplicate / {duplicate:true}，不会落第二条（契约 v0.56 §3.13）。
|
||||
val network = listOf(
|
||||
CyreneError.LocalKind.NoNetwork,
|
||||
CyreneError.LocalKind.Dns,
|
||||
@@ -195,29 +205,66 @@ class SendFallbackTest {
|
||||
CyreneError.LocalKind.ConnectFailed,
|
||||
)
|
||||
network.forEach { kind ->
|
||||
val error = CyreneError.Local(kind)
|
||||
assertEquals(
|
||||
"$kind 属于网络类失败，该进队列",
|
||||
RestFailureLanding.Queue,
|
||||
restFailureLanding(CyreneError.Local(kind)),
|
||||
RestAttemptLanding.Unsent,
|
||||
restAttemptLanding(Result.failure(CyreneException(error))),
|
||||
)
|
||||
// 同一档在 W41 那套落点判据上也是队列（口径一字未改，[restAttemptLanding] 只是复用它）。
|
||||
assertEquals(RestFailureLanding.Queue, restFailureLanding(error))
|
||||
}
|
||||
}
|
||||
|
||||
@Test
|
||||
fun `a fallback that got no verdict at all queues`() {
|
||||
// `null` = 本端自己设的那道上限到点了（withTimeoutOrNull 回 null）或抛了个映射不出来的
|
||||
// 异常：服务端没给过裁决 ⇒ 这条「还没上路」，与网络类失败同一档。
|
||||
fun `no verdict at all queues, never a hopeful success`() {
|
||||
// `null` = 本端自己设的整通上限到点了（withTimeoutOrNull 回 null）：服务端没给过裁决。
|
||||
// 这条可能已经被服务端收下（只是回执没回来），也可能压根没到——本端分不出来，而**分不出来
|
||||
// 就往队列倒**（fail-closed）。这是「一条没上路的消息绝不被当成发出去了」的落点。
|
||||
assertEquals(RestAttemptLanding.Unsent, restAttemptLanding(null))
|
||||
assertEquals(RestFailureLanding.Queue, restFailureLanding(null))
|
||||
|
||||
// 异常映射不出来（不是 CyreneException）同理：本端说不清服务端有没有收下 ⇒ 队列。
|
||||
assertEquals(
|
||||
RestAttemptLanding.Unsent,
|
||||
restAttemptLanding(Result.failure<ChatAck>(IllegalStateException("boom"))),
|
||||
)
|
||||
}
|
||||
|
||||
@Test
|
||||
fun `a verdict from the server is never re-sent behind the user's back`() {
|
||||
// 服务端**回了话**就是它的裁决：拒绝（4xx/5xx 的明确码）与未授权都不该被悄悄重发——
|
||||
// 重发换不来不同结果，而且把「服务端拒绝」说成「还没连上」是假话。
|
||||
fun `isAccepted covers every ack shape without a default`() {
|
||||
// 逐一穷举回执面：受理族（起回合 / 续接排队 / 幂等重复）为真，拒绝族（队列满 / 接不住 /
|
||||
// 明确拒绝）为假。`ChatAck` 将来多一档时这里会编译不过——必须有人来做决定。
|
||||
assertTrue(ChatAck.Accepted.isAccepted())
|
||||
assertTrue(ChatAck.Queued(queueDepth = 0).isAccepted())
|
||||
assertTrue(ChatAck.Duplicate(messageId = null, sessionId = null).isAccepted())
|
||||
|
||||
assertTrue(!ChatAck.QueueFull.isAccepted())
|
||||
assertTrue(!ChatAck.Unavailable("chat_unavailable", "").isAccepted())
|
||||
assertTrue(!ChatAck.Rejected("bad_request", "").isAccepted())
|
||||
}
|
||||
|
||||
@Test
|
||||
fun `every local kind is decided, none is inherited by accident`() {
|
||||
// 穷举 `LocalKind`：要么进队列、要么回滚，**没有第三种默认**。这条断言的价值在错误模型将来
|
||||
// 多一档时——那时 `restFailureLanding` 会编译不过（when 不穷举），而不是静默把新档位当成
|
||||
// 「网络类」重发出去。
|
||||
CyreneError.LocalKind.entries.forEach { kind ->
|
||||
val landing = restFailureLanding(CyreneError.Local(kind))
|
||||
assertTrue(
|
||||
"$kind 必须被明确分到某一档",
|
||||
landing == RestFailureLanding.Queue || landing == RestFailureLanding.Rollback,
|
||||
)
|
||||
}
|
||||
assertEquals(
|
||||
RestFailureLanding.Rollback,
|
||||
restFailureLanding(CyreneError.Server("bad_request", "invalid text", 400)),
|
||||
RestFailureLanding.Queue,
|
||||
restFailureLanding(CyreneError.Local(CyreneError.LocalKind.NoNetwork)),
|
||||
)
|
||||
assertNotEquals(
|
||||
RestFailureLanding.Queue,
|
||||
restFailureLanding(CyreneError.Local(CyreneError.LocalKind.Tls)),
|
||||
)
|
||||
// 服务端裁决的那两类：也都回滚（[CyreneError.Server] 有 code/status，[CyreneError.Unauthorized] 没有）。
|
||||
assertEquals(
|
||||
RestFailureLanding.Rollback,
|
||||
restFailureLanding(CyreneError.Server("internal", "boom", 500)),
|
||||
@@ -228,58 +275,40 @@ class SendFallbackTest {
|
||||
)
|
||||
}
|
||||
|
||||
@Test
|
||||
fun `configuration failures roll back because waiting will not fix them`() {
|
||||
// 配置与协议这一族要的是用户改设置 / 换地址，等下去不会变好；回滚时那句 userMessage
|
||||
// 正是他需要看到的东西（把「证书不对」说成「还没连上」会让他干等）。
|
||||
val config = listOf(
|
||||
CyreneError.LocalKind.Tls,
|
||||
CyreneError.LocalKind.CleartextBlocked,
|
||||
CyreneError.LocalKind.InvalidUrl,
|
||||
CyreneError.LocalKind.Parse,
|
||||
CyreneError.LocalKind.Unknown,
|
||||
)
|
||||
config.forEach { kind ->
|
||||
assertEquals(
|
||||
"$kind 等下去不会变好，该回滚并说清",
|
||||
RestFailureLanding.Rollback,
|
||||
restFailureLanding(CyreneError.Local(kind)),
|
||||
)
|
||||
}
|
||||
}
|
||||
// —— 三、两个超时（无网快速进队列的关键） ——
|
||||
|
||||
@Test
|
||||
fun `every local kind is decided, none is inherited by accident`() {
|
||||
// 穷举 `LocalKind`：判据要么进队列、要么回滚，**没有第三种默认**。这条断言的价值在
|
||||
// 错误模型将来多一档时——那时 `restFailureLanding` 会编译不过（when 不穷举），
|
||||
// 而不是静默把新档位当成「网络类」重发出去。
|
||||
CyreneError.LocalKind.entries.forEach { kind ->
|
||||
val landing = restFailureLanding(CyreneError.Local(kind))
|
||||
assertTrue(
|
||||
"$kind 必须被明确分到某一档",
|
||||
landing == RestFailureLanding.Queue || landing == RestFailureLanding.Rollback,
|
||||
)
|
||||
}
|
||||
// 队列那一档恰好是「请求没被服务端读到」的网络族；其余一律回滚。
|
||||
assertEquals(
|
||||
RestFailureLanding.Queue,
|
||||
restFailureLanding(CyreneError.Local(CyreneError.LocalKind.NoNetwork)),
|
||||
fun `the connect timeout is short enough to sit inside a send`() {
|
||||
// 硬要求：**完全无网时消息要在 ~3 秒内进队列**。W45 起这条性质由 client 上的**连接超时**
|
||||
// 承担（不再有探针）：连不上就在这个量级内失败 ⇒ 进队列。既不能长到像卡死，也不能短到正常
|
||||
// 网络下误连不上。它直接加在用户的发送动作上（气泡已经上屏、用户等着这条消息的结论）。
|
||||
assertTrue(
|
||||
"连接超时 $CONNECT_TIMEOUT_MILLIS ms 落在 2~3 秒之外",
|
||||
CONNECT_TIMEOUT_MILLIS in 2_000L..3_000L,
|
||||
)
|
||||
assertNotEquals(
|
||||
RestFailureLanding.Queue,
|
||||
restFailureLanding(CyreneError.Local(CyreneError.LocalKind.Tls)),
|
||||
|
||||
// 第二个超时（整通上限）另设、且**必须比连接超时宽**：它管的是「连上了但一直不回话」那一档
|
||||
// （半死的代理、接住连接却不回应的服务端），到点也按网络类失败 ⇒ 进队列。
|
||||
assertTrue(
|
||||
"整通上限该比连接超时宽裕",
|
||||
REST_FALLBACK_TIMEOUT_MILLIS > CONNECT_TIMEOUT_MILLIS,
|
||||
)
|
||||
// 且**不许拉长**：它是用户按完发送键之后的干等时间上界，W41 定在 10 秒，本值就是 10 秒。
|
||||
assertTrue(
|
||||
"整通上限不许超过 W41 定的 10 秒，实际 $REST_FALLBACK_TIMEOUT_MILLIS ms",
|
||||
REST_FALLBACK_TIMEOUT_MILLIS <= 10_000L,
|
||||
)
|
||||
}
|
||||
|
||||
// —— 五、这条回落到不了「REST 专属」的消息 ——
|
||||
// —— 四、这条阶梯到不了「REST 专属」的消息 ——
|
||||
|
||||
@Test
|
||||
fun `image and document messages never enter this ladder at all`() {
|
||||
// 回落 REST 只服务「本该走 WS 的短文本」：有图 / 有附件的消息在**进入阶梯之前**就被
|
||||
// 这条阶梯只服务「本该走 WS 的短文本」：有图 / 有附件的消息在**进入阶梯之前**就被
|
||||
// [ChatSendRouting.channelFor] 判成 Rest，走的是另一条分支（今天的 `sendViaRest` /
|
||||
// `deliverImage`，行为一个字没变）。所以它们既不会因为「WS 未就绪」而多花一次探测，
|
||||
// 也不会因为回落失败而进队列——**没有本地 URI 过一阵子还能不能读、图片要不要重编码
|
||||
// 这一类新问题**（队列不持久化，真跨了进程 URI 授权就失效）。
|
||||
// `deliverImage`，行为一个字没变）。所以它们既不会因为「WS 未就绪」而被试一次 REST，
|
||||
// 也不会因为试败而进队列——**没有本地 URI 过一阵子还能不能读、图片要不要重编码这一类
|
||||
// 新问题**（队列不持久化，真跨了进程 URI 授权就失效）。
|
||||
// 路由规则本身由 `ChatSendPolicyTest` 逐个钉住，这里只钉「阶梯的入口因此变窄」这件事。
|
||||
val documents = listOf(DraftDocument("content://doc", "a.txt"))
|
||||
|
||||
|
||||
Reference in new issue
Block a user