fix(processing): 落地 D1/D5/D6 三项裁决，删除 head-deadline 参数
D5（终态判据只保留尝试上限）： - Pump.tick 内联 attempts 判定，删除 head-deadline 相关的毒丸分支与滞留告警代码 - 删除配置项 head-deadline（PipelineProps / application.yml）与 PumpDeadlineTest - PROCESSING_STARTED_AT 变为只写，注释如实说明当前无判据消费它 D1（回填放弃判据改为时间）： - 暂时性故障在 R 之前只退避重试，不再按尝试次数放弃；到 R 才放弃并记 TRANSIENT_DEADLINE - backfill-max-attempts 降级为单行重试的告警阈值 D6（超期判据改用本地入队时间）： - 新增 V6 迁移：PROC_STATE 加 ENQUEUED_AT（回填存量后置为非空 + 默认） - findBackfillDue 的谓词与 overdue 标记改比较 enqueued_at，不再用库方时钟的 received_at - BackfillDue 增加 overdue；收报与兼容入口显式写入本地入队时间 文档同步： - 清理 4 处 message-lifecycle.md 章节号死链（Pump/InboxService/PipelineProps/application.yml） - 关闭 G-HEAD-DEADLINE、G-BACKFILL-ABANDON-BYTIME、G-ENQUEUED-AT 三条缺口登记 - reference/user-stories/README 与实现对齐 验证：./gradlew test ⇒ 122 tests, 0 failures, 1 skipped Refs: ACM2-45
This commit is contained in:
windyboy
@@ -26,7 +26,6 @@ class PipelineProps {
|
||||
// 因此最多只用得到 max-attempts − 1 个档位。多出来的档位永远走不到（会被 validate() 拦下）。
|
||||
var backoffMs: List<Long> = listOf(1000, 2000, 4000, 8000)
|
||||
var backoffCapMs: Long = 60_000
|
||||
var headDeadline: Duration = Duration.ofMinutes(10) // 最坏 HOL 上界（毒丸升级）
|
||||
|
||||
/**
|
||||
* 缺口等待时长：水位后面缺了一个 ID 时，等这么久还没出现就认定它永远不会来了，
|
||||
@@ -42,7 +41,7 @@ class PipelineProps {
|
||||
* （比如回填一直失败），扫描谓词的超期分支成立，无视退避强制补写（只会提前、从不推迟打标）。
|
||||
*
|
||||
* 唯一约束是 R ≤ R_keep，**不保护重放窗口**（打标时刻与 R 解耦，完整论证见
|
||||
* message-lifecycle.md §5.2）。取值待 Q6 定案，确认之前不要为提速下调。
|
||||
* design「回填」）。取值待 Q6 定案，确认之前不要为提速下调。
|
||||
*/
|
||||
var overdueBackfill: Duration = Duration.ofDays(30)
|
||||
|
||||
|
||||
@@ -41,7 +41,7 @@ enum class ErrorClass {
|
||||
/** 解码逻辑的问题；修好 codec 之后可以重放。 */
|
||||
CODEC_ERROR,
|
||||
|
||||
/** 重试次数用尽或队头滞留超时；人工复核后可以重放。 */
|
||||
/** 重试次数用尽（尝试上限）；人工复核后可以重放。 */
|
||||
EXHAUSTED,
|
||||
|
||||
/** 数据库、网络等基础设施抖动，重试通常就能过。 */
|
||||
@@ -64,8 +64,13 @@ data class ProcState(
|
||||
val errorClass: ErrorClass? = null,
|
||||
/** 最近一次失败的原因（截断后落库，供排查）。 */
|
||||
val lastError: String? = null,
|
||||
/** 信箱里的接收时间：用来判断"超期仍未回填"，也是最老未处理信龄的计算依据。 */
|
||||
/** 信箱里的接收时间，来自**库方时钟**、**可能为 NULL**；仅用于对账与展示，不作任何判据。 */
|
||||
val receivedAt: Instant? = null,
|
||||
/**
|
||||
* 本地入队时间（本系统写入，非空）。超期补写期限 `R` **只比较它**：与判据用的本地
|
||||
* `NOW` 同源，不受库方时钟偏斜影响（`PRE-4`）。
|
||||
*/
|
||||
val enqueuedAt: Instant? = null,
|
||||
/** 非空表示已确认信箱行带上了处理标记。 */
|
||||
val backfillAt: Instant? = null,
|
||||
/** 非空表示还欠一次回填：写终态时置为当前时间，失败后退避推后。 */
|
||||
@@ -73,15 +78,20 @@ data class ProcState(
|
||||
val backfillAttempts: Int = 0,
|
||||
val backfillError: String? = null,
|
||||
/**
|
||||
* 非空表示已判定"不必再回填"（信箱行不存在，或达到尝试上限）。
|
||||
* 非空表示已判定"不必再回填"：信箱行不存在，或暂时性故障持续到 `R` 仍未打标。
|
||||
*
|
||||
* **它不等于标记已确认**：`backfillAt` 仍为空，所以不满足"边界内全部行已打标"的清除条件。
|
||||
* 停止重试与"已满足清除前提"是两件事，不能互相替代。
|
||||
*/
|
||||
val backfillAbandonedAt: Instant? = null,
|
||||
/** 放弃原因（`MISSING_ROW` / `MAX_ATTEMPTS`），供人工对账与恢复判断。 */
|
||||
/** 放弃原因（`MISSING_ROW` / `TRANSIENT_DEADLINE`），供人工对账与恢复判断。 */
|
||||
val backfillAbandonedReason: String? = null,
|
||||
/** 首次被主泵取得的时刻；重试不刷新，用作 HOL deadline 的稳定起点。 */
|
||||
/**
|
||||
* 首次被主泵取得的时刻；重试不刷新。
|
||||
*
|
||||
* 当前没有判据消费它（终态只看尝试上限），保留作排障与后续扩展；写入方是
|
||||
* [com.gzzn.omms.msgexchange.infra.persistence.ProcStateRepository.markProcessingStartedIfAbsent]。
|
||||
*/
|
||||
val processingStartedAt: Instant? = null,
|
||||
val updatedAt: Instant = Instant.now(),
|
||||
)
|
||||
|
||||
@@ -45,10 +45,11 @@ interface ProcStateRepository {
|
||||
*
|
||||
* 幂等：同一个消息 ID 重复登记既不报错、也不会建第二行（收报重扫和兼容入口并发调用都安全）。
|
||||
*
|
||||
* @param receivedAt 信箱里的接收时间，用于判断超期未回填和统计最老信龄
|
||||
* @param receivedAt 信箱里的接收时间，来自库方时钟、可能为 NULL；只作对账与展示
|
||||
* @param enqueuedAt 本地入队时间；不传时由实现用自己注入的时钟填（列非空）
|
||||
* @return true 表示这次真的新建了一行
|
||||
*/
|
||||
fun insertIfAbsent(msgId: Long, receivedAt: Instant?): Boolean
|
||||
fun insertIfAbsent(msgId: Long, receivedAt: Instant?, enqueuedAt: Instant? = null): Boolean
|
||||
|
||||
fun find(msgId: Long): ProcState?
|
||||
|
||||
@@ -99,11 +100,13 @@ interface ProcStateRepository {
|
||||
fun recordBackfillFailure(msgId: Long, error: String?, attempts: Int, nextAttemptAt: Instant, now: Instant)
|
||||
|
||||
/**
|
||||
* 放弃回填：判定该行不必再自动尝试（信箱行不存在，或达到尝试上限）。
|
||||
* 放弃回填：判定该行不必再自动尝试（信箱行不存在，或暂时性故障持续到 `R` 仍未打标）。
|
||||
*
|
||||
* **abandoned ≠ 标记已确认**：`BACKFILL_AT` 仍为空，因此**不**满足"边界内全部行已打标"的
|
||||
* 清除前提；放弃只是停止自动重试并把事实留痕，供人工对账。
|
||||
*
|
||||
* 放弃判据是**时间**（`R` 超期）而不是尝试次数：一次小时级的共享库故障不该把待回填行成批判死。
|
||||
*
|
||||
* @return false 表示该行不存在
|
||||
*/
|
||||
fun markBackfillAbandoned(msgId: Long, reason: String, now: Instant): Boolean
|
||||
@@ -117,8 +120,9 @@ interface ProcStateRepository {
|
||||
/**
|
||||
* 找出现在该回填的记录：已经到终态、还没确认回填，并且退避时间已到。
|
||||
*
|
||||
* [overdueBefore] 是兜底：消息接收时间早于它的（已经等了很久）无视退避直接补写。
|
||||
* [overdueBefore] 是兜底：入队时间早于它的（已经等了超过 `R`）无视退避直接补写。
|
||||
* 没有这条兜底，退避一直失败的话这些行就永远打不上标记，库方也没法清理信箱。
|
||||
* 判据比较的是**本地** `ENQUEUED_AT`，不是库方时钟的 `RECEIVED_AT`（`PRE-4`）。
|
||||
*/
|
||||
fun findBackfillDue(now: Instant, overdueBefore: Instant, limit: Int): List<BackfillDue>
|
||||
|
||||
@@ -132,8 +136,12 @@ interface ProcStateRepository {
|
||||
fun hasAny(): Boolean
|
||||
}
|
||||
|
||||
/** 扫描到的待回填记录。 */
|
||||
data class BackfillDue(val msgId: Long, val attempts: Int)
|
||||
/**
|
||||
* 扫描到的待回填记录。
|
||||
* @param overdue 入队时间早于 `NOW − R`：本轮已经进入强补写窗口
|
||||
* （调用方据此决定"暂时性故障是否到放弃期限"）
|
||||
*/
|
||||
data class BackfillDue(val msgId: Long, val attempts: Int, val overdue: Boolean = false)
|
||||
|
||||
/**
|
||||
* 处理侧积压快照。
|
||||
|
||||
+24
-12
@@ -76,16 +76,20 @@ class JdbcProcStateRepository(
|
||||
private val clock: Clock,
|
||||
) : ProcStateRepository {
|
||||
/** 入队（幂等）：主键冲突时什么都不做，所以重复扫描和兼容入口并发调用都安全。 */
|
||||
override fun insertIfAbsent(msgId: Long, receivedAt: Instant?): Boolean =
|
||||
ds.update(
|
||||
"INSERT INTO proc_state (msg_id, state, received_at, updated_at) VALUES (?, 'PENDING', ?, ?) " +
|
||||
"ON CONFLICT (msg_id) DO NOTHING",
|
||||
override fun insertIfAbsent(msgId: Long, receivedAt: Instant?, enqueuedAt: Instant?): Boolean {
|
||||
val now = clock.instant()
|
||||
return ds.update(
|
||||
"INSERT INTO proc_state (msg_id, state, received_at, enqueued_at, updated_at) " +
|
||||
"VALUES (?, 'PENDING', ?, ?, ?) ON CONFLICT (msg_id) DO NOTHING",
|
||||
{ ps ->
|
||||
ps.setLong(1, msgId)
|
||||
ps.setTimestamp(2, receivedAt?.toSqlTimestamp())
|
||||
ps.setTimestamp(3, clock.instant().toSqlTimestamp())
|
||||
// 入队时间必须由本系统时钟给出：它同时是超期判据 R 的比较对象（PRE-4）。
|
||||
ps.setTimestamp(3, (enqueuedAt ?: now).toSqlTimestamp())
|
||||
ps.setTimestamp(4, now.toSqlTimestamp())
|
||||
},
|
||||
) == 1
|
||||
}
|
||||
|
||||
override fun find(msgId: Long): ProcState? =
|
||||
ds.queryOne("$SELECT_PROC WHERE msg_id = ?", { ps -> ps.setLong(1, msgId) }, ::mapProcState)
|
||||
@@ -254,22 +258,29 @@ class JdbcProcStateRepository(
|
||||
* 排序用**公平轮转**：先按已尝试次数升序，再按 msg_id。若只按 msg_id 升序，
|
||||
* 最旧的一批永久失败行会持续占满批次，后面的记录永远轮不到（全局回填饥饿）。
|
||||
*/
|
||||
/**
|
||||
* 超期判据用 `enqueued_at`（本地时钟、非空），不用 `received_at`：
|
||||
* 后者来自库方时钟，前偏会让 `NOW − R` 提前成立（`PRE-4`）。
|
||||
*/
|
||||
override fun findBackfillDue(now: Instant, overdueBefore: Instant, limit: Int): List<BackfillDue> =
|
||||
ds.query(
|
||||
"""
|
||||
SELECT msg_id, backfill_attempts FROM proc_state
|
||||
SELECT msg_id, backfill_attempts, (enqueued_at < ?) AS overdue FROM proc_state
|
||||
WHERE backfill_at IS NULL
|
||||
AND backfill_abandoned_at IS NULL
|
||||
AND state IN ('SUCCEEDED', 'SKIPPED', 'DEAD')
|
||||
AND (backfill_next_at IS NULL OR backfill_next_at <= ? OR (received_at IS NOT NULL AND received_at < ?))
|
||||
AND (backfill_next_at IS NULL OR backfill_next_at <= ? OR enqueued_at < ?)
|
||||
ORDER BY backfill_attempts ASC, msg_id ASC LIMIT ?
|
||||
""".trimIndent(),
|
||||
{ ps ->
|
||||
ps.setTimestamp(1, now.toSqlTimestamp())
|
||||
ps.setTimestamp(2, overdueBefore.toSqlTimestamp())
|
||||
ps.setInt(3, limit)
|
||||
ps.setTimestamp(1, overdueBefore.toSqlTimestamp())
|
||||
ps.setTimestamp(2, now.toSqlTimestamp())
|
||||
ps.setTimestamp(3, overdueBefore.toSqlTimestamp())
|
||||
ps.setInt(4, limit)
|
||||
},
|
||||
) { rs -> BackfillDue(rs.getLong("msg_id"), rs.getInt("backfill_attempts")) }
|
||||
) { rs ->
|
||||
BackfillDue(rs.getLong("msg_id"), rs.getInt("backfill_attempts"), rs.getBoolean("overdue"))
|
||||
}
|
||||
|
||||
override fun requeueByErrorClasses(errorClasses: List<ErrorClass>): Int {
|
||||
if (errorClasses.isEmpty()) return 0
|
||||
@@ -321,6 +332,7 @@ class JdbcProcStateRepository(
|
||||
errorClass = rs.getString("error_class")?.let(ErrorClass::valueOf),
|
||||
lastError = rs.getString("last_error"),
|
||||
receivedAt = rs.getInstant("received_at"),
|
||||
enqueuedAt = rs.getInstant("enqueued_at"),
|
||||
backfillAt = rs.getInstant("backfill_at"),
|
||||
backfillNextAt = rs.getInstant("backfill_next_at"),
|
||||
backfillAttempts = rs.getInt("backfill_attempts"),
|
||||
@@ -334,7 +346,7 @@ class JdbcProcStateRepository(
|
||||
private companion object {
|
||||
const val SELECT_PROC =
|
||||
"SELECT msg_id, state, identity_key, attempts, next_attempt_at, error_class, last_error, " +
|
||||
"received_at, backfill_at, backfill_next_at, backfill_attempts, backfill_error, " +
|
||||
"received_at, enqueued_at, backfill_at, backfill_next_at, backfill_attempts, backfill_error, " +
|
||||
"backfill_abandoned_at, backfill_abandoned_reason, processing_started_at, updated_at FROM proc_state"
|
||||
}
|
||||
}
|
||||
|
||||
@@ -63,9 +63,17 @@ class StubProcState : ProcStateRepository {
|
||||
rows.clear(); bound.clear()
|
||||
}
|
||||
|
||||
override fun insertIfAbsent(msgId: Long, receivedAt: Instant?): Boolean {
|
||||
/**
|
||||
* 内存版没有独立时钟，入队时间默认沿用 `receivedAt`（测试可显式传入以区分两者）；
|
||||
* 生产实现由注入的 Clock 填，`ENQUEUED_AT` 列非空。
|
||||
*/
|
||||
override fun insertIfAbsent(msgId: Long, receivedAt: Instant?, enqueuedAt: Instant?): Boolean {
|
||||
if (rows.containsKey(msgId)) return false
|
||||
rows[msgId] = ProcState(msgId, ProcStatus.PENDING, receivedAt = receivedAt)
|
||||
rows[msgId] = ProcState(
|
||||
msgId, ProcStatus.PENDING,
|
||||
receivedAt = receivedAt,
|
||||
enqueuedAt = enqueuedAt ?: receivedAt,
|
||||
)
|
||||
return true
|
||||
}
|
||||
|
||||
@@ -182,14 +190,13 @@ class StubProcState : ProcStateRepository {
|
||||
override fun findBackfillDue(now: Instant, overdueBefore: Instant, limit: Int): List<BackfillDue> =
|
||||
rows.values
|
||||
.filter { it.state.isTerminal() && it.backfillAt == null && it.backfillAbandonedAt == null }
|
||||
.filter {
|
||||
it.backfillNextAt == null || it.backfillNextAt <= now ||
|
||||
(it.receivedAt != null && it.receivedAt < overdueBefore)
|
||||
}
|
||||
// 超期判据是本地入队时间，与 JDBC 实现保持一致（不用库方时钟的 receivedAt）。
|
||||
.map { it to (it.enqueuedAt?.let { e -> e < overdueBefore } ?: false) }
|
||||
.filter { (row, overdue) -> overdue || row.backfillNextAt == null || row.backfillNextAt <= now }
|
||||
// 公平轮转：先按已尝试次数，再按 msg_id。只按 msg_id 会让最旧的一批永久失败行占满批次。
|
||||
.sortedWith(compareBy({ it.backfillAttempts }, { it.msgId }))
|
||||
.sortedWith(compareBy({ it.first.backfillAttempts }, { it.first.msgId }))
|
||||
.take(limit)
|
||||
.map { BackfillDue(it.msgId, it.backfillAttempts) }
|
||||
.map { (row, overdue) -> BackfillDue(row.msgId, row.backfillAttempts, overdue) }
|
||||
|
||||
override fun hasAny(): Boolean = rows.isNotEmpty()
|
||||
|
||||
|
||||
@@ -77,7 +77,8 @@ class InboxPoller(
|
||||
val enqueued = txManager.inTransaction {
|
||||
var n = 0
|
||||
rows.takeWhile { it.msgId <= committedTo }.forEach { row ->
|
||||
if (procState.insertIfAbsent(row.msgId, row.receivedAt)) n++
|
||||
// 入队时间用本轮注入的本地时钟，与超期判据 R 同源（PRE-4）。
|
||||
if (procState.insertIfAbsent(row.msgId, row.receivedAt, enqueuedAt = now)) n++
|
||||
}
|
||||
cursor.save(InboxCursorRepository.Cursor(committedTo, holeSince))
|
||||
n
|
||||
|
||||
@@ -14,7 +14,7 @@ import java.time.Instant
|
||||
* 收报轮询会按 ID 把它补进来，所以不会丢消息。
|
||||
*
|
||||
* 该入口**不参与水位**：它直接写 PROC_STATE，登记的行可能超出水位；主泵只领 `msgId ≤ W`，
|
||||
* 因此不破坏 FIFO——这类行等水位追平后按序自然领取（message-lifecycle.md §5.1）。
|
||||
* 因此不破坏 FIFO——这类行等水位追平后按序自然领取（`invariants.md` INV-4）。
|
||||
*/
|
||||
@Singleton
|
||||
class InboxService(
|
||||
@@ -35,8 +35,10 @@ class InboxService(
|
||||
if (mailboxReceivedAt == null) {
|
||||
log.warn("mailbox receive time missing msgId={}, falling back to enqueue time", id)
|
||||
}
|
||||
val receivedAt = mailboxReceivedAt ?: clock.instant()
|
||||
procState.insertIfAbsent(id, receivedAt)
|
||||
val now = clock.instant()
|
||||
val receivedAt = mailboxReceivedAt ?: now
|
||||
// 入队时间单独传本地时钟：它是超期判据 R 的比较对象，不能借用库方时间（PRE-4）。
|
||||
procState.insertIfAbsent(id, receivedAt, enqueuedAt = now)
|
||||
log.info("compat-accepted msgId={}", id)
|
||||
return Receipt(id, receivedAt)
|
||||
}
|
||||
|
||||
@@ -19,9 +19,12 @@ import java.time.Instant
|
||||
* 这个类负责第二件：
|
||||
*
|
||||
* - [attempt]：单行写标记，只由 [sweep] 逐行调用——主泵与处理器不直接调它（回填一律扫描驱动）。
|
||||
* - [sweep]：定时把还欠回填的记录挑出来重试。失败就按 30 秒起步、最长 15 分钟的
|
||||
* 退避往后推；如果一条消息从收到现在已经超过超期期限，则无视退避强制补写——
|
||||
* 否则退避可能一直失败下去，这些行永远打不上标记，库方就没法清理信箱。
|
||||
* - [sweep]：定时把还欠回填的记录挑出来重试。失败按 30 秒起步、最长 15 分钟的退避往后推；
|
||||
* 入队时间早于 `NOW − R` 的行无视退避、每轮都试。
|
||||
*
|
||||
* 放弃判据是**时间**（`R` 超期）而不是尝试次数：一次小时级的共享库故障不该把待回填行成批
|
||||
* 判死、再要求人工成批恢复。`R` 之前只退避重试；到 `R` 仍未打标才停止自动重试并进放弃清单
|
||||
* （保留 [reopen] 人工恢复）。`backfill-max-attempts` 因此降级为单行重试的告警阈值。
|
||||
*
|
||||
* 两条底线：回填失败不会把终态改回去，也不会重新执行业务逻辑；写标记只写还是空标记的
|
||||
* 行，重复执行没有副作用。
|
||||
@@ -46,8 +49,8 @@ class BackfillService(
|
||||
/** 放弃原因：运行时查询确认信箱行不存在（确定性结论，重试不会改变结果）。 */
|
||||
const val ABANDON_MISSING_ROW = "MISSING_ROW"
|
||||
|
||||
/** 放弃原因：暂时性故障达到尝试上限；停止自动重试，但保留人工恢复能力。 */
|
||||
const val ABANDON_MAX_ATTEMPTS = "MAX_ATTEMPTS"
|
||||
/** 放弃原因：暂时性故障持续到 `R` 仍未打标；停止自动重试，但保留人工恢复能力。 */
|
||||
const val ABANDON_TRANSIENT_DEADLINE = "TRANSIENT_DEADLINE"
|
||||
|
||||
fun backoffDelayFor(attempts: Int): Duration {
|
||||
val shift = (attempts - 1).coerceIn(0, 20)
|
||||
@@ -59,13 +62,17 @@ class BackfillService(
|
||||
* 处理完立刻试一次。失败只记一笔退避信息就返回，不抛异常——
|
||||
* 调用方是主泵的处理路径，不能被回填问题拖住。
|
||||
*/
|
||||
fun attempt(msgId: Long, now: Instant = clock.instant()) {
|
||||
/**
|
||||
* @param overdue 该行入队时间早于 `NOW − R`（已进入强补写窗口）。通常来自 [sweep] 的扫描结果；
|
||||
* 直接调用默认 false，即按普通退避处理。
|
||||
*/
|
||||
fun attempt(msgId: Long, overdue: Boolean = false, now: Instant = clock.instant()) {
|
||||
lifecycleGate.exclusive {
|
||||
val row = procState.find(msgId) ?: return@exclusive
|
||||
if (row.state !in TERMINAL_STATES) return@exclusive
|
||||
if (row.backfillAt != null) return@exclusive
|
||||
if (row.backfillAbandonedAt != null) return@exclusive
|
||||
record(msgId, attempts = row.backfillAttempts, now = now)?.let {
|
||||
record(msgId, attempts = row.backfillAttempts, overdue = overdue, now = now)?.let {
|
||||
log.warn("backfill failed msgId={} error={} (sweep will retry)", msgId, it)
|
||||
}
|
||||
}
|
||||
@@ -89,12 +96,12 @@ class BackfillService(
|
||||
*/
|
||||
fun sweep(now: Instant = clock.instant()): Int {
|
||||
val due = procState.findBackfillDue(now, now.minus(props.pipeline.overdueBackfill), props.pipeline.backfillBatch)
|
||||
due.forEach { attempt(it.msgId, now) }
|
||||
due.forEach { attempt(it.msgId, it.overdue, now) }
|
||||
return due.size
|
||||
}
|
||||
|
||||
/** 回填一条。@return 失败原因；返回 null 表示已处理完（写成功／早已标记／已放弃）。 */
|
||||
private fun record(msgId: Long, attempts: Int, now: Instant): String? =
|
||||
private fun record(msgId: Long, attempts: Int, overdue: Boolean, now: Instant): String? =
|
||||
try {
|
||||
when (mailbox.markProcessedIfUnmarked(msgId, mailboxProps.processedValue)) {
|
||||
MailboxMarkResult.MARKED, MailboxMarkResult.ALREADY_MARKED -> {
|
||||
@@ -111,18 +118,25 @@ class BackfillService(
|
||||
}
|
||||
}
|
||||
} catch (e: Exception) {
|
||||
// 超时/连接失败是暂时性的，**不能**当作缺行证据：按退避重试。
|
||||
// 达到上限后停止自动重试（保留人工恢复能力），避免永久占满扫描批次造成饥饿。
|
||||
// 超时/连接失败是暂时性的，**不能**当作缺行证据，也不能按次数放弃：
|
||||
// 按退避重试，直到入队时间超过 R 才停止自动重试（保留人工恢复能力）。
|
||||
val reason = e.message ?: e.javaClass.simpleName
|
||||
val nextAttempts = attempts + 1
|
||||
if (nextAttempts >= props.pipeline.backfillMaxAttempts) {
|
||||
runCatching { procState.markBackfillAbandoned(msgId, ABANDON_MAX_ATTEMPTS, now) }
|
||||
if (overdue) {
|
||||
runCatching { procState.markBackfillAbandoned(msgId, ABANDON_TRANSIENT_DEADLINE, now) }
|
||||
.onFailure { log.error("abandon backfill failed msgId={}", msgId, it) }
|
||||
log.error("backfill abandoned after {} attempts msgId={} error={}", nextAttempts, msgId, reason)
|
||||
log.error("backfill abandoned at R msgId={} attempts={} error={}", msgId, nextAttempts, reason)
|
||||
} else {
|
||||
runCatching {
|
||||
procState.recordBackfillFailure(msgId, reason, nextAttempts, now.plus(backoffDelayFor(nextAttempts)), now)
|
||||
}.onFailure { log.error("record backfill failure failed msgId={}", msgId, it) }
|
||||
if (nextAttempts >= props.pipeline.backfillMaxAttempts) {
|
||||
log.warn(
|
||||
"backfill retries for msgId={} reached the warning threshold ({}); " +
|
||||
"still retrying until R elapses — 放弃判据是 R 超期，不是次数",
|
||||
msgId, props.pipeline.backfillMaxAttempts,
|
||||
)
|
||||
}
|
||||
}
|
||||
reason
|
||||
}
|
||||
|
||||
@@ -25,8 +25,7 @@ import java.util.concurrent.atomic.AtomicLong
|
||||
*
|
||||
* 每次 tick 只看当前最小的未完成消息（"队头"）：
|
||||
* - 没有待处理消息就睡一个轮询间隔；
|
||||
* - 队头失败了还在退避期，就等到能重试的时刻；如果重试次数用尽或滞留太久，
|
||||
* 直接转死信，不放任它一直堵着；
|
||||
* - 队头失败了还在退避期，就等到能重试的时刻；重试次数用尽才转死信，不放任它一直堵着；
|
||||
* - 其余情况交给 [MessageProcessor] 处理。
|
||||
*
|
||||
* 一次只处理一条是刻意的。后面的消息不能越过卡住的队头，否则同一条航班的报文
|
||||
@@ -81,38 +80,36 @@ class Pump(
|
||||
//
|
||||
// 水位以内的行都是收报按 ID 顺序发现并登记的；水位之外的行只可能来自兼容入口
|
||||
// 直接写 PROC_STATE（它不参与水位）。若允许领取，它就会越过那些尚未入队的较小 ID，
|
||||
// 破坏 FIFO（不变量"只领取已发现的行"，message-lifecycle.md §11）。这种行在空洞补齐、`W` 追平之后自然可领取。
|
||||
// 破坏 FIFO（不变量"只领取已发现的行"，`invariants.md` INV-4）。这种行在空洞补齐、`W` 追平之后自然可领取。
|
||||
val watermark = cursor.load().committedUpTo
|
||||
if (head.msgId > watermark) {
|
||||
warnBeyondWatermark(head.msgId, watermark)
|
||||
sleepQuietly(props.pipeline.pollInterval)
|
||||
return
|
||||
}
|
||||
val now = clock.instant()
|
||||
when {
|
||||
head.state == ProcStatus.FAILED && poisoned(head) -> {
|
||||
head.state == ProcStatus.FAILED && head.attempts >= props.pipeline.maxAttempts -> {
|
||||
log.error("poison -> DEAD msgId={} attempts={} lastError={}", head.msgId, head.attempts, head.lastError)
|
||||
// markTerminal 在同一条 UPDATE 里登记回填意图；回填由扫描补写，不在这里做跨库写。
|
||||
procState.markTerminal(
|
||||
head.msgId, ProcStatus.DEAD,
|
||||
errorClass = ErrorClass.EXHAUSTED,
|
||||
lastError = head.lastError ?: "head-deadline-exceeded",
|
||||
lastError = head.lastError ?: "attempts-exhausted",
|
||||
attempts = head.attempts,
|
||||
now = clock.instant(),
|
||||
now = now,
|
||||
)
|
||||
}
|
||||
head.state == ProcStatus.FAILED && (head.nextAttemptAt ?: Instant.EPOCH) > clock.instant() ->
|
||||
sleepQuietly(Duration.between(clock.instant(), head.nextAttemptAt))
|
||||
head.state == ProcStatus.FAILED && (head.nextAttemptAt ?: Instant.EPOCH) > now ->
|
||||
sleepQuietly(Duration.between(now, head.nextAttemptAt))
|
||||
// 其余情况（新消息，或退避到期的重试）交给处理入口
|
||||
else -> {
|
||||
procState.markProcessingStartedIfAbsent(head.msgId, clock.instant())
|
||||
procState.markProcessingStartedIfAbsent(head.msgId, now)
|
||||
processor.processOne(head)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
private fun poisoned(head: ProcState): Boolean =
|
||||
isHeadPoisoned(head, clock.instant(), props)
|
||||
|
||||
/** 上一次"队头在水位之外"告警时的水位值：只在它变化时告警，避免每秒刷屏。 */
|
||||
private val warnedWatermark = AtomicLong(Long.MIN_VALUE)
|
||||
|
||||
@@ -131,10 +128,6 @@ class Pump(
|
||||
}
|
||||
}
|
||||
|
||||
internal fun isHeadPoisoned(head: ProcState, now: Instant, props: PipelineProps): Boolean =
|
||||
head.attempts >= props.pipeline.maxAttempts ||
|
||||
Duration.between(head.processingStartedAt ?: head.updatedAt, now) >= props.pipeline.headDeadline
|
||||
|
||||
/**
|
||||
* 处理一条消息：读原文 → 解码 → 绑定业务身份 → 分派给对应处理器。
|
||||
*
|
||||
|
||||
@@ -18,11 +18,10 @@ msgx:
|
||||
max-attempts: 5 # 处理/投递同值
|
||||
backoff-ms: [1000, 2000, 4000, 8000] # 指数退避；档位数必须 = max-attempts - 1（启动自检）
|
||||
backoff-cap-ms: 60000
|
||||
head-deadline: 10m # 队头滞留上界 = 最坏 HOL 时长（毒丸升级）
|
||||
max-commit-delay: 5m # §5.1 空洞老化：W+1 空洞超过该时延判定为永久（Q2 最大提交时延）
|
||||
overdue-backfill: 30d # §5.2 超期补写期限 R（Q6）：仅须 R ≤ R_keep，不保护重放窗口（message-lifecycle §5.2）
|
||||
overdue-backfill: 30d # 超期补写期限 R（Q6）：仅须 R ≤ R_keep，不保护重放窗口；判据比较本地 ENQUEUED_AT（design「回填」）
|
||||
backfill-batch: 100 # 回填扫描单批条数
|
||||
backfill-max-attempts: 100 # 回填自动重试上限；达上限停止自动重试（可人工恢复），避免占满批次
|
||||
backfill-max-attempts: 100 # 单行重试的告警阈值；放弃判据是 R 超期，不是次数（design「回填」）
|
||||
# 一次性切流播种：默认（注释掉）不播种。min=读现存全部 | zero=从 0 按空洞规则 | max=跳过可见存量 | <id>
|
||||
# cutover-watermark: min
|
||||
late-detect-period: 60s # 迟到到达检测（只读，阶段 0）：复查被放行的空洞 ID 是否后来真的出现；0=关闭
|
||||
|
||||
@@ -0,0 +1,22 @@
|
||||
-- =====================================================================
|
||||
-- V6：PROC_STATE 增加本地入队时间 ENQUEUED_AT
|
||||
-- ---------------------------------------------------------------------
|
||||
-- 只动自有 PostgreSQL；共享 MySQL 不建表、不改结构。
|
||||
--
|
||||
-- 背景（门禁裁决 D6）：超期补写期限 `R` 原先比较 `RECEIVED_AT`，那是**库方写入的异地时钟**。
|
||||
-- 库方时钟前偏时 `RECEIVED_AT < NOW − R` 会提前成立，在「打标即可清除」语义下正好制造
|
||||
-- 原文被提前清除的丢失窗口（`PRE-4`）。改为比较本系统自己写入的 `ENQUEUED_AT`：
|
||||
-- 本地、非空、与判据用的 `NOW` 同源。`RECEIVED_AT` 退回「仅用于对账与展示」，并允许为 NULL。
|
||||
--
|
||||
-- 存量行没有历史入队时间，只能一次性尽力回填：优先信箱接收时间，其次最后一次更新时间。
|
||||
-- 这是一次推断，不回写业务、不影响任何已有终态，也不改已发布的迁移历史。
|
||||
-- =====================================================================
|
||||
|
||||
ALTER TABLE PROC_STATE ADD COLUMN ENQUEUED_AT TIMESTAMP(6) WITH TIME ZONE;
|
||||
|
||||
UPDATE PROC_STATE
|
||||
SET ENQUEUED_AT = COALESCE(RECEIVED_AT, UPDATED_AT, now())
|
||||
WHERE ENQUEUED_AT IS NULL;
|
||||
|
||||
ALTER TABLE PROC_STATE ALTER COLUMN ENQUEUED_AT SET NOT NULL;
|
||||
ALTER TABLE PROC_STATE ALTER COLUMN ENQUEUED_AT SET DEFAULT now();
|
||||
+14
-4
@@ -83,10 +83,10 @@ class FlywayMigrationTest {
|
||||
assertEquals(setOf("flid", "operation_day", "state", "state_version", "last_msg_id"), cols)
|
||||
}
|
||||
|
||||
// 回填相关的列都落在 PROC_STATE 上（收信时间判断超期；abandoned 记录"停止自动重试"）
|
||||
// 回填相关的列都落在 PROC_STATE 上（本地入队时间判断超期；abandoned 记录"停止自动重试"）
|
||||
stmt.executeQuery(
|
||||
"SELECT column_name FROM information_schema.columns WHERE table_name = 'proc_state' " +
|
||||
"AND column_name IN ('received_at', 'backfill_at', 'backfill_next_at', " +
|
||||
"AND column_name IN ('received_at', 'enqueued_at', 'backfill_at', 'backfill_next_at', " +
|
||||
"'backfill_attempts', 'backfill_error', 'backfill_abandoned_at', " +
|
||||
"'backfill_abandoned_reason', 'processing_started_at')",
|
||||
).use { rs ->
|
||||
@@ -94,13 +94,23 @@ class FlywayMigrationTest {
|
||||
while (rs.next()) cols.add(rs.getString("column_name"))
|
||||
assertEquals(
|
||||
setOf(
|
||||
"received_at", "backfill_at", "backfill_next_at", "backfill_attempts", "backfill_error",
|
||||
"backfill_abandoned_at", "backfill_abandoned_reason", "processing_started_at",
|
||||
"received_at", "enqueued_at", "backfill_at", "backfill_next_at", "backfill_attempts",
|
||||
"backfill_error", "backfill_abandoned_at", "backfill_abandoned_reason",
|
||||
"processing_started_at",
|
||||
),
|
||||
cols,
|
||||
)
|
||||
}
|
||||
|
||||
// V6：入队时间是超期判据 R 的比较对象，必须非空（received_at 则允许为 NULL）
|
||||
stmt.executeQuery(
|
||||
"SELECT is_nullable FROM information_schema.columns " +
|
||||
"WHERE table_name = 'proc_state' AND column_name = 'enqueued_at'",
|
||||
).use { rs ->
|
||||
assertTrue(rs.next(), "V6 必须已加上 ENQUEUED_AT 列")
|
||||
assertEquals("NO", rs.getString("is_nullable"), "ENQUEUED_AT 必须非空")
|
||||
}
|
||||
|
||||
// 两张单行表（管道锁、收报水位）的种子数据都要在
|
||||
stmt.executeQuery("SELECT count(*) FROM pipeline_lock WHERE lock_id = 1").use { rs ->
|
||||
assertTrue(rs.next())
|
||||
|
||||
+12
-5
@@ -100,7 +100,7 @@ class InboxLifecycleJdbcSqlTest {
|
||||
seed(2L, t0)
|
||||
proc.markTerminal(2L, ProcStatus.SUCCEEDED, now = t0)
|
||||
proc.recordBackfillFailure(2L, "mysql-down", 1, t0.plus(Duration.ofMinutes(15)), t0)
|
||||
// ③ 退避还没到，但收信时间已经很久了：应当无视退避直接补写
|
||||
// ③ 退避还没到，但本地入队时间已经超过 R：应当无视退避直接补写
|
||||
seed(3L, overdue)
|
||||
proc.markTerminal(3L, ProcStatus.DEAD, errorClass = ErrorClass.MALFORMED, now = t0)
|
||||
proc.recordBackfillFailure(3L, "mysql-down", 1, t0.plus(Duration.ofMinutes(15)), t0)
|
||||
@@ -111,10 +111,13 @@ class InboxLifecycleJdbcSqlTest {
|
||||
proc.markTerminal(5L, ProcStatus.SUCCEEDED, now = t0)
|
||||
proc.markBackfilled(5L, t0)
|
||||
|
||||
val due = proc.findBackfillDue(t0, t0.minus(Duration.ofDays(30)), limit = 100).map { it.msgId }
|
||||
val due = proc.findBackfillDue(t0, t0.minus(Duration.ofDays(30)), limit = 100)
|
||||
val ids = due.map { it.msgId }
|
||||
|
||||
assertEquals(listOf(1L, 3L), due)
|
||||
assertEquals(listOf(1L, 3L), ids)
|
||||
assertEquals(listOf(1L), proc.findBackfillDue(t0, t0.minus(Duration.ofDays(30)), limit = 1).map { it.msgId })
|
||||
// overdue 标记只在入队时间早于 NOW − R 时为真；调用方据此决定暂时性故障是否到放弃期限。
|
||||
assertEquals(listOf(1L to false, 3L to true), due.map { it.msgId to it.overdue })
|
||||
}
|
||||
|
||||
@Test
|
||||
@@ -186,15 +189,18 @@ class InboxLifecycleJdbcSqlTest {
|
||||
assertEquals(MailboxMarkResult.MISSING, mailbox.markProcessedIfUnmarked(999L, "PROCESSED"))
|
||||
}
|
||||
|
||||
/** 播种一行：`enqueued_at` 与 `received_at` 取同一时刻（超期判据只看前者）。 */
|
||||
private fun seed(msgId: Long, receivedAt: Instant?, state: ProcStatus = ProcStatus.PENDING) {
|
||||
ds.connection.use { conn: Connection ->
|
||||
conn.prepareStatement(
|
||||
"INSERT INTO proc_state (msg_id, state, received_at, backfill_attempts, updated_at) VALUES (?, ?, ?, 0, ?)",
|
||||
"INSERT INTO proc_state (msg_id, state, received_at, enqueued_at, backfill_attempts, updated_at) " +
|
||||
"VALUES (?, ?, ?, ?, 0, ?)",
|
||||
).use { ps ->
|
||||
ps.setLong(1, msgId)
|
||||
ps.setString(2, state.name)
|
||||
ps.setTimestamp(3, receivedAt?.let { Timestamp.from(it) })
|
||||
ps.setTimestamp(4, Timestamp.from(t0))
|
||||
ps.setTimestamp(4, receivedAt?.let { Timestamp.from(it) } ?: Timestamp.from(t0))
|
||||
ps.setTimestamp(5, Timestamp.from(t0))
|
||||
ps.executeUpdate()
|
||||
}
|
||||
}
|
||||
@@ -219,6 +225,7 @@ class InboxLifecycleJdbcSqlTest {
|
||||
error_class VARCHAR(20),
|
||||
last_error VARCHAR(1000),
|
||||
received_at TIMESTAMP WITH TIME ZONE,
|
||||
enqueued_at TIMESTAMP WITH TIME ZONE NOT NULL,
|
||||
backfill_at TIMESTAMP WITH TIME ZONE,
|
||||
backfill_next_at TIMESTAMP WITH TIME ZONE,
|
||||
backfill_attempts INT NOT NULL DEFAULT 0,
|
||||
|
||||
@@ -26,7 +26,11 @@ class ReplayServiceTest {
|
||||
)
|
||||
}
|
||||
|
||||
override fun insertIfAbsent(msgId: Long, receivedAt: Instant?): Boolean = rows.putIfAbsent(msgId, ProcState(msgId, ProcStatus.PENDING, receivedAt = receivedAt)) == null
|
||||
override fun insertIfAbsent(msgId: Long, receivedAt: Instant?, enqueuedAt: Instant?): Boolean =
|
||||
rows.putIfAbsent(
|
||||
msgId,
|
||||
ProcState(msgId, ProcStatus.PENDING, receivedAt = receivedAt, enqueuedAt = enqueuedAt ?: receivedAt),
|
||||
) == null
|
||||
override fun find(msgId: Long): ProcState? = rows[msgId]
|
||||
override fun findSuccessTerminal(msgId: Long): Boolean = rows[msgId]?.state == ProcStatus.SUCCEEDED
|
||||
override fun headUnfinished(): ProcState? = null
|
||||
|
||||
@@ -301,25 +301,44 @@ class BackfillServiceTest {
|
||||
assertNotNull(proc.find(911L)!!.backfillNextAt)
|
||||
}
|
||||
|
||||
/**
|
||||
* 门禁裁决 D1：暂时性故障**不按次数放弃**。即使超过 `backfill-max-attempts` 这个
|
||||
* 告警阈值，也继续退避重试——一次小时级的共享库故障不该把待回填行成批判死。
|
||||
*/
|
||||
@Test
|
||||
fun `transient failures keep retrying until the attempt cap and stay recoverable`() {
|
||||
fun `transient failures keep retrying past the warning threshold`() {
|
||||
props.pipeline.backfillMaxAttempts = 3
|
||||
val proc = StubProcState()
|
||||
val mailbox = FakeMailbox(fail = true)
|
||||
succeeded(proc, 912L)
|
||||
val svc = service(proc, mailbox)
|
||||
|
||||
svc.attempt(912L) // 1 次：暂时性故障，只退避
|
||||
svc.attempt(912L) // 2 次
|
||||
assertNull(proc.find(912L)!!.backfillAbandonedAt)
|
||||
repeat(5) { svc.attempt(912L) }
|
||||
|
||||
svc.attempt(912L) // 3 次：达到上限 → 停止自动重试
|
||||
val row = proc.find(912L)!!
|
||||
assertEquals(BackfillService.ABANDON_MAX_ATTEMPTS, row.backfillAbandonedReason)
|
||||
assertEquals(5, row.backfillAttempts)
|
||||
assertNull(row.backfillAbandonedAt) // 次数不是放弃判据
|
||||
assertNull(row.backfillAt)
|
||||
assertNotNull(row.backfillNextAt) // 仍在退避重试
|
||||
}
|
||||
|
||||
assertTrue(svc.reopen(912L)) // 人工恢复入口存在且有效
|
||||
assertNull(proc.find(912L)!!.backfillAbandonedAt)
|
||||
/** 到 `R` 仍未打标（overdue）时，暂时性故障才停止自动重试并进放弃清单。 */
|
||||
@Test
|
||||
fun `overdue transient failure is abandoned with the deadline reason and stays recoverable`() {
|
||||
val proc = StubProcState()
|
||||
val mailbox = FakeMailbox(fail = true)
|
||||
succeeded(proc, 913L)
|
||||
val svc = service(proc, mailbox)
|
||||
|
||||
svc.attempt(913L, overdue = true)
|
||||
|
||||
val row = proc.find(913L)!!
|
||||
assertEquals(BackfillService.ABANDON_TRANSIENT_DEADLINE, row.backfillAbandonedReason)
|
||||
assertNotNull(row.backfillAbandonedAt)
|
||||
assertNull(row.backfillAt) // 放弃 ≠ 标记已确认
|
||||
|
||||
assertTrue(svc.reopen(913L)) // 人工恢复入口存在且有效
|
||||
assertNull(proc.find(913L)!!.backfillAbandonedAt)
|
||||
}
|
||||
|
||||
@Test
|
||||
@@ -346,18 +365,20 @@ class BackfillServiceTest {
|
||||
}
|
||||
|
||||
/**
|
||||
* 语义固定（G4）：`RECEIVED_AT` 为 NULL 时"超期"分支不成立，`R` 兜底**不生效**，
|
||||
* 该行只能靠退避重试。把这条钉住，避免以后误以为 `R` 一定能兜底。
|
||||
* 语义固定（门禁裁决 D6）：超期判据是**本地入队时间** `ENQUEUED_AT`，与 `RECEIVED_AT` 无关。
|
||||
* 上游没给接收时间（NULL）不再让 `R` 兜底失效——这正是引入 `ENQUEUED_AT` 要消除的窗口。
|
||||
*/
|
||||
@Test
|
||||
fun `null received time disables the overdue shortcut so only backoff applies`() {
|
||||
fun `null received time no longer disables the overdue shortcut`() {
|
||||
val proc = StubProcState()
|
||||
proc.insertIfAbsent(921L, null) // 上游未提供接收时间
|
||||
val old = t0.minus(props.pipeline.overdueBackfill).minusSeconds(60)
|
||||
proc.insertIfAbsent(921L, null, enqueuedAt = old) // 上游未提供接收时间
|
||||
proc.markTerminal(921L, ProcStatus.SUCCEEDED, now = t0)
|
||||
proc.recordBackfillFailure(921L, "mysql-down", attempts = 1, nextAttemptAt = t0.plusSeconds(3600), now = t0)
|
||||
|
||||
val due = proc.findBackfillDue(now = t0, overdueBefore = t0.plusSeconds(10_000), limit = 10)
|
||||
val due = proc.findBackfillDue(now = t0, overdueBefore = t0.minus(props.pipeline.overdueBackfill), limit = 10)
|
||||
|
||||
assertTrue(due.isEmpty()) // 超期分支无效 + 退避未到期
|
||||
assertEquals(listOf(921L), due.map { it.msgId }) // 退避未到期，但入队时间已超 R → 仍被扫描
|
||||
assertTrue(due.single().overdue)
|
||||
}
|
||||
}
|
||||
|
||||
@@ -1,40 +0,0 @@
|
||||
package com.gzzn.omms.msgexchange.processing
|
||||
|
||||
import com.gzzn.omms.msgexchange.config.PipelineProps
|
||||
import com.gzzn.omms.msgexchange.domain.ProcState
|
||||
import com.gzzn.omms.msgexchange.domain.ProcStatus
|
||||
import org.junit.jupiter.api.Assertions.assertFalse
|
||||
import org.junit.jupiter.api.Assertions.assertTrue
|
||||
import org.junit.jupiter.api.Test
|
||||
import java.time.Instant
|
||||
|
||||
class PumpDeadlineTest {
|
||||
private val props = PipelineProps()
|
||||
private val started = Instant.parse("2026-09-08T03:00:00Z")
|
||||
|
||||
@Test
|
||||
fun `deadline uses stable processing start rather than refreshed update time`() {
|
||||
val row = ProcState(
|
||||
msgId = 1,
|
||||
state = ProcStatus.FAILED,
|
||||
attempts = 1,
|
||||
processingStartedAt = started,
|
||||
updatedAt = started.plusSeconds(590),
|
||||
)
|
||||
|
||||
assertTrue(isHeadPoisoned(row, started.plusSeconds(600), props))
|
||||
}
|
||||
|
||||
@Test
|
||||
fun `head below attempt and time limits remains retryable`() {
|
||||
val row = ProcState(
|
||||
msgId = 1,
|
||||
state = ProcStatus.FAILED,
|
||||
attempts = 1,
|
||||
processingStartedAt = started,
|
||||
updatedAt = started.plusSeconds(590),
|
||||
)
|
||||
|
||||
assertFalse(isHeadPoisoned(row, started.plusSeconds(599), props))
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user