refactor(processing): remove unused processing start field
processing_started_at 没有任何判据消费(终态只看尝试上限),删除字段、仓储接口与方法、主泵写入、重放清理、SELECT/映射与 H2 fixture,并新增 V7 DROP COLUMN;V1–V6 历史迁移不动。 验证:./gradlew test 123 tests / 0 fail;V7 的 DROP 与最终 schema 断言需真实 PG,本机无 Docker 时 FlywayMigrationTest 按既有 assumeTrue 跳过。
This commit is contained in:
@@ -86,12 +86,5 @@ data class ProcState(
|
|||||||
val backfillAbandonedAt: Instant? = null,
|
val backfillAbandonedAt: Instant? = null,
|
||||||
/** 放弃原因(`MISSING_ROW` / `TRANSIENT_DEADLINE`),供人工对账与恢复判断。 */
|
/** 放弃原因(`MISSING_ROW` / `TRANSIENT_DEADLINE`),供人工对账与恢复判断。 */
|
||||||
val backfillAbandonedReason: String? = null,
|
val backfillAbandonedReason: String? = null,
|
||||||
/**
|
|
||||||
* 首次被主泵取得的时刻;重试不刷新。
|
|
||||||
*
|
|
||||||
* 当前没有判据消费它(终态只看尝试上限),保留作排障与后续扩展;写入方是
|
|
||||||
* [com.gzzn.omms.msgexchange.infra.persistence.ProcStateRepository.markProcessingStartedIfAbsent]。
|
|
||||||
*/
|
|
||||||
val processingStartedAt: Instant? = null,
|
|
||||||
val updatedAt: Instant = Instant.now(),
|
val updatedAt: Instant = Instant.now(),
|
||||||
)
|
)
|
||||||
|
|||||||
@@ -59,9 +59,6 @@ interface ProcStateRepository {
|
|||||||
/** 当前队头:还没处理完的消息里 ID 最小的那条。 */
|
/** 当前队头:还没处理完的消息里 ID 最小的那条。 */
|
||||||
fun headUnfinished(): ProcState?
|
fun headUnfinished(): ProcState?
|
||||||
|
|
||||||
/** 首次开始处理时记下稳定起点;重试不覆盖。 */
|
|
||||||
fun markProcessingStartedIfAbsent(msgId: Long, now: Instant)
|
|
||||||
|
|
||||||
/** 给消息绑定业务身份;返回 false 表示这个身份已经被另一条消息占了(业务重复)。 */
|
/** 给消息绑定业务身份;返回 false 表示这个身份已经被另一条消息占了(业务重复)。 */
|
||||||
fun tryBindIdentity(msgId: Long, identityKey: String): Boolean
|
fun tryBindIdentity(msgId: Long, identityKey: String): Boolean
|
||||||
|
|
||||||
|
|||||||
+2
-13
@@ -107,16 +107,6 @@ class JdbcProcStateRepository(
|
|||||||
::mapProcState,
|
::mapProcState,
|
||||||
)
|
)
|
||||||
|
|
||||||
override fun markProcessingStartedIfAbsent(msgId: Long, now: Instant) {
|
|
||||||
ds.update(
|
|
||||||
"UPDATE proc_state SET processing_started_at = ? WHERE msg_id = ? AND processing_started_at IS NULL",
|
|
||||||
{ ps ->
|
|
||||||
ps.setTimestamp(1, now.toSqlTimestamp())
|
|
||||||
ps.setLong(2, msgId)
|
|
||||||
},
|
|
||||||
)
|
|
||||||
}
|
|
||||||
|
|
||||||
override fun tryBindIdentity(msgId: Long, identityKey: String): Boolean {
|
override fun tryBindIdentity(msgId: Long, identityKey: String): Boolean {
|
||||||
ownerOfIdentity(identityKey)?.let { owner ->
|
ownerOfIdentity(identityKey)?.let { owner ->
|
||||||
return owner == msgId
|
return owner == msgId
|
||||||
@@ -286,7 +276,7 @@ class JdbcProcStateRepository(
|
|||||||
if (errorClasses.isEmpty()) return 0
|
if (errorClasses.isEmpty()) return 0
|
||||||
val placeholders = errorClasses.joinToString(",") { "?" }
|
val placeholders = errorClasses.joinToString(",") { "?" }
|
||||||
return ds.update(
|
return ds.update(
|
||||||
"UPDATE proc_state SET state = 'PENDING', attempts = 0, next_attempt_at = NULL, processing_started_at = NULL, updated_at = ? " +
|
"UPDATE proc_state SET state = 'PENDING', attempts = 0, next_attempt_at = NULL, updated_at = ? " +
|
||||||
"WHERE state IN ('FAILED', 'DEAD') AND error_class IN ($placeholders)",
|
"WHERE state IN ('FAILED', 'DEAD') AND error_class IN ($placeholders)",
|
||||||
{ ps ->
|
{ ps ->
|
||||||
ps.setTimestamp(1, clock.instant().toSqlTimestamp())
|
ps.setTimestamp(1, clock.instant().toSqlTimestamp())
|
||||||
@@ -339,7 +329,6 @@ class JdbcProcStateRepository(
|
|||||||
backfillError = rs.getString("backfill_error"),
|
backfillError = rs.getString("backfill_error"),
|
||||||
backfillAbandonedAt = rs.getInstant("backfill_abandoned_at"),
|
backfillAbandonedAt = rs.getInstant("backfill_abandoned_at"),
|
||||||
backfillAbandonedReason = rs.getString("backfill_abandoned_reason"),
|
backfillAbandonedReason = rs.getString("backfill_abandoned_reason"),
|
||||||
processingStartedAt = rs.getInstant("processing_started_at"),
|
|
||||||
updatedAt = rs.getInstant("updated_at") ?: clock.instant(),
|
updatedAt = rs.getInstant("updated_at") ?: clock.instant(),
|
||||||
)
|
)
|
||||||
|
|
||||||
@@ -347,7 +336,7 @@ class JdbcProcStateRepository(
|
|||||||
const val SELECT_PROC =
|
const val SELECT_PROC =
|
||||||
"SELECT msg_id, state, identity_key, attempts, next_attempt_at, error_class, last_error, " +
|
"SELECT msg_id, state, identity_key, attempts, next_attempt_at, error_class, last_error, " +
|
||||||
"received_at, enqueued_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"
|
"backfill_abandoned_at, backfill_abandoned_reason, updated_at FROM proc_state"
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
|||||||
@@ -84,12 +84,6 @@ class StubProcState : ProcStateRepository {
|
|||||||
override fun headUnfinished(): ProcState? =
|
override fun headUnfinished(): ProcState? =
|
||||||
rows.values.filter { it.state == ProcStatus.PENDING || it.state == ProcStatus.FAILED }.minByOrNull { it.msgId }
|
rows.values.filter { it.state == ProcStatus.PENDING || it.state == ProcStatus.FAILED }.minByOrNull { it.msgId }
|
||||||
|
|
||||||
override fun markProcessingStartedIfAbsent(msgId: Long, now: Instant) {
|
|
||||||
rows[msgId]?.let { row ->
|
|
||||||
if (row.processingStartedAt == null) rows[msgId] = row.copy(processingStartedAt = now)
|
|
||||||
}
|
|
||||||
}
|
|
||||||
|
|
||||||
override fun tryBindIdentity(msgId: Long, identityKey: String): Boolean {
|
override fun tryBindIdentity(msgId: Long, identityKey: String): Boolean {
|
||||||
val owner = bound[identityKey]
|
val owner = bound[identityKey]
|
||||||
if (owner != null && owner != msgId) return false
|
if (owner != null && owner != msgId) return false
|
||||||
@@ -216,7 +210,7 @@ class StubProcState : ProcStateRepository {
|
|||||||
var n = 0
|
var n = 0
|
||||||
rows.forEach { (id, s) ->
|
rows.forEach { (id, s) ->
|
||||||
if (s.errorClass in errorClasses && (s.state == ProcStatus.FAILED || s.state == ProcStatus.DEAD)) {
|
if (s.errorClass in errorClasses && (s.state == ProcStatus.FAILED || s.state == ProcStatus.DEAD)) {
|
||||||
rows[id] = s.copy(state = ProcStatus.PENDING, attempts = 0, nextAttemptAt = null, processingStartedAt = null)
|
rows[id] = s.copy(state = ProcStatus.PENDING, attempts = 0, nextAttemptAt = null)
|
||||||
n++
|
n++
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|||||||
@@ -103,10 +103,7 @@ class Pump(
|
|||||||
head.state == ProcStatus.FAILED && (head.nextAttemptAt ?: Instant.EPOCH) > now ->
|
head.state == ProcStatus.FAILED && (head.nextAttemptAt ?: Instant.EPOCH) > now ->
|
||||||
sleepQuietly(Duration.between(now, head.nextAttemptAt))
|
sleepQuietly(Duration.between(now, head.nextAttemptAt))
|
||||||
// 其余情况(新消息,或退避到期的重试)交给处理入口
|
// 其余情况(新消息,或退避到期的重试)交给处理入口
|
||||||
else -> {
|
else -> processor.processOne(head)
|
||||||
procState.markProcessingStartedIfAbsent(head.msgId, now)
|
|
||||||
processor.processOne(head)
|
|
||||||
}
|
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
|||||||
@@ -0,0 +1,14 @@
|
|||||||
|
-- =====================================================================
|
||||||
|
-- V7:PROC_STATE 删除 PROCESSING_STARTED_AT
|
||||||
|
-- ---------------------------------------------------------------------
|
||||||
|
-- 只动自有 PostgreSQL;共享 MySQL 不建表、不改结构。
|
||||||
|
--
|
||||||
|
-- 该列由 V3 引入(原 HOL deadline 的稳定处理起点),但代码里没有任何判据消费它:
|
||||||
|
-- 处理终态只看尝试上限(`PARAM:msgx.pipeline.max-attempts`)。按精简原则删除列与
|
||||||
|
-- 全部读写路径,不留"以后也许有用"的死字段;将来若为 `CLM-9` 需要处理开始时间,
|
||||||
|
-- 先写文档再实现。
|
||||||
|
--
|
||||||
|
-- 已发布的迁移历史(V1–V6)保持不变。
|
||||||
|
-- =====================================================================
|
||||||
|
|
||||||
|
ALTER TABLE PROC_STATE DROP COLUMN PROCESSING_STARTED_AT;
|
||||||
+15
-3
@@ -48,7 +48,7 @@ class FlywayMigrationTest {
|
|||||||
while (rs.next()) {
|
while (rs.next()) {
|
||||||
records.add(Triple(rs.getString("version"), rs.getString("script"), rs.getBoolean("success")))
|
records.add(Triple(rs.getString("version"), rs.getString("script"), rs.getBoolean("success")))
|
||||||
}
|
}
|
||||||
assertTrue(records.size >= 5, "flyway_schema_history must record all migrations")
|
assertTrue(records.size >= 7, "flyway_schema_history must record all migrations")
|
||||||
assertEquals("1", records[0].first)
|
assertEquals("1", records[0].first)
|
||||||
assertEquals("V1__flight_state_baseline.sql", records[0].second)
|
assertEquals("V1__flight_state_baseline.sql", records[0].second)
|
||||||
assertEquals("2", records[1].first)
|
assertEquals("2", records[1].first)
|
||||||
@@ -59,6 +59,10 @@ class FlywayMigrationTest {
|
|||||||
assertEquals("V4__backfill_closure.sql", records[3].second)
|
assertEquals("V4__backfill_closure.sql", records[3].second)
|
||||||
assertEquals("5", records[4].first)
|
assertEquals("5", records[4].first)
|
||||||
assertEquals("V5__cutover_seed.sql", records[4].second)
|
assertEquals("V5__cutover_seed.sql", records[4].second)
|
||||||
|
assertEquals("6", records[5].first)
|
||||||
|
assertEquals("V6__enqueued_at.sql", records[5].second)
|
||||||
|
assertEquals("7", records[6].first)
|
||||||
|
assertEquals("V7__drop_processing_started_at.sql", records[6].second)
|
||||||
assertTrue(records.all { it.third })
|
assertTrue(records.all { it.third })
|
||||||
}
|
}
|
||||||
|
|
||||||
@@ -88,7 +92,7 @@ class FlywayMigrationTest {
|
|||||||
"SELECT column_name FROM information_schema.columns WHERE table_name = 'proc_state' " +
|
"SELECT column_name FROM information_schema.columns WHERE table_name = 'proc_state' " +
|
||||||
"AND column_name IN ('received_at', 'enqueued_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_attempts', 'backfill_error', 'backfill_abandoned_at', " +
|
||||||
"'backfill_abandoned_reason', 'processing_started_at')",
|
"'backfill_abandoned_reason')",
|
||||||
).use { rs ->
|
).use { rs ->
|
||||||
val cols = mutableSetOf<String>()
|
val cols = mutableSetOf<String>()
|
||||||
while (rs.next()) cols.add(rs.getString("column_name"))
|
while (rs.next()) cols.add(rs.getString("column_name"))
|
||||||
@@ -96,12 +100,20 @@ class FlywayMigrationTest {
|
|||||||
setOf(
|
setOf(
|
||||||
"received_at", "enqueued_at", "backfill_at", "backfill_next_at", "backfill_attempts",
|
"received_at", "enqueued_at", "backfill_at", "backfill_next_at", "backfill_attempts",
|
||||||
"backfill_error", "backfill_abandoned_at", "backfill_abandoned_reason",
|
"backfill_error", "backfill_abandoned_at", "backfill_abandoned_reason",
|
||||||
"processing_started_at",
|
|
||||||
),
|
),
|
||||||
cols,
|
cols,
|
||||||
)
|
)
|
||||||
}
|
}
|
||||||
|
|
||||||
|
// V7:PROC_STATE 不再保留处理开始时间(无判据消费)
|
||||||
|
stmt.executeQuery(
|
||||||
|
"SELECT count(*) FROM information_schema.columns WHERE table_name = 'proc_state' " +
|
||||||
|
"AND column_name = 'processing_started_at'",
|
||||||
|
).use { rs ->
|
||||||
|
assertTrue(rs.next())
|
||||||
|
assertEquals(0, rs.getInt(1), "V7 必须已删除 PROCESSING_STARTED_AT")
|
||||||
|
}
|
||||||
|
|
||||||
// V6:入队时间是超期判据 R 的比较对象,必须非空(received_at 则允许为 NULL)
|
// V6:入队时间是超期判据 R 的比较对象,必须非空(received_at 则允许为 NULL)
|
||||||
stmt.executeQuery(
|
stmt.executeQuery(
|
||||||
"SELECT is_nullable FROM information_schema.columns " +
|
"SELECT is_nullable FROM information_schema.columns " +
|
||||||
|
|||||||
-10
@@ -142,15 +142,6 @@ class InboxLifecycleJdbcSqlTest {
|
|||||||
assertEquals(InboxCursorRepository.Cursor(42L, t0), cursor.load())
|
assertEquals(InboxCursorRepository.Cursor(42L, t0), cursor.load())
|
||||||
}
|
}
|
||||||
|
|
||||||
@Test
|
|
||||||
fun `processing start is written once and is not refreshed`() {
|
|
||||||
seed(11L, t0)
|
|
||||||
proc.markProcessingStartedIfAbsent(11L, t0.plusSeconds(10))
|
|
||||||
proc.markProcessingStartedIfAbsent(11L, t0.plusSeconds(20))
|
|
||||||
|
|
||||||
assertEquals(t0.plusSeconds(10), proc.find(11L)!!.processingStartedAt)
|
|
||||||
}
|
|
||||||
|
|
||||||
@Test
|
@Test
|
||||||
fun `pipeline transaction rolls back terminal and outbox writes after a late failure`() {
|
fun `pipeline transaction rolls back terminal and outbox writes after a late failure`() {
|
||||||
seed(11L, t0)
|
seed(11L, t0)
|
||||||
@@ -232,7 +223,6 @@ class InboxLifecycleJdbcSqlTest {
|
|||||||
backfill_error VARCHAR(512),
|
backfill_error VARCHAR(512),
|
||||||
backfill_abandoned_at TIMESTAMP WITH TIME ZONE,
|
backfill_abandoned_at TIMESTAMP WITH TIME ZONE,
|
||||||
backfill_abandoned_reason VARCHAR(64),
|
backfill_abandoned_reason VARCHAR(64),
|
||||||
processing_started_at TIMESTAMP WITH TIME ZONE,
|
|
||||||
updated_at TIMESTAMP WITH TIME ZONE NOT NULL,
|
updated_at TIMESTAMP WITH TIME ZONE NOT NULL,
|
||||||
CONSTRAINT uk_proc_identity UNIQUE (identity_key)
|
CONSTRAINT uk_proc_identity UNIQUE (identity_key)
|
||||||
)
|
)
|
||||||
|
|||||||
@@ -22,7 +22,6 @@ class ReplayServiceTest {
|
|||||||
fun seed(id: Long, status: ProcStatus, ec: ErrorClass?) {
|
fun seed(id: Long, status: ProcStatus, ec: ErrorClass?) {
|
||||||
rows[id] = ProcState(
|
rows[id] = ProcState(
|
||||||
id, status, identityKey = "k$id", attempts = 3, errorClass = ec, lastError = "x",
|
id, status, identityKey = "k$id", attempts = 3, errorClass = ec, lastError = "x",
|
||||||
processingStartedAt = Instant.parse("2026-09-08T03:00:00Z"),
|
|
||||||
)
|
)
|
||||||
}
|
}
|
||||||
|
|
||||||
@@ -34,7 +33,6 @@ class ReplayServiceTest {
|
|||||||
override fun find(msgId: Long): ProcState? = rows[msgId]
|
override fun find(msgId: Long): ProcState? = rows[msgId]
|
||||||
override fun findSuccessTerminal(msgId: Long): Boolean = rows[msgId]?.state == ProcStatus.SUCCEEDED
|
override fun findSuccessTerminal(msgId: Long): Boolean = rows[msgId]?.state == ProcStatus.SUCCEEDED
|
||||||
override fun headUnfinished(): ProcState? = null
|
override fun headUnfinished(): ProcState? = null
|
||||||
override fun markProcessingStartedIfAbsent(msgId: Long, now: Instant) = Unit
|
|
||||||
override fun tryBindIdentity(msgId: Long, identityKey: String): Boolean = true
|
override fun tryBindIdentity(msgId: Long, identityKey: String): Boolean = true
|
||||||
override fun ownerOfIdentity(identityKey: String): Long? = null
|
override fun ownerOfIdentity(identityKey: String): Long? = null
|
||||||
override fun update(
|
override fun update(
|
||||||
@@ -69,7 +67,7 @@ class ReplayServiceTest {
|
|||||||
val s = rows[id]!!
|
val s = rows[id]!!
|
||||||
if (s.errorClass != null && s.errorClass in errorClasses &&
|
if (s.errorClass != null && s.errorClass in errorClasses &&
|
||||||
(s.state == ProcStatus.FAILED || s.state == ProcStatus.DEAD)) {
|
(s.state == ProcStatus.FAILED || s.state == ProcStatus.DEAD)) {
|
||||||
rows[id] = s.copy(state = ProcStatus.PENDING, attempts = 0, nextAttemptAt = null, processingStartedAt = null)
|
rows[id] = s.copy(state = ProcStatus.PENDING, attempts = 0, nextAttemptAt = null)
|
||||||
n++
|
n++
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
@@ -92,7 +90,6 @@ class ReplayServiceTest {
|
|||||||
assertEquals(ProcStatus.PENDING, repo.rows[1]!!.state)
|
assertEquals(ProcStatus.PENDING, repo.rows[1]!!.state)
|
||||||
assertEquals(0, repo.rows[1]!!.attempts)
|
assertEquals(0, repo.rows[1]!!.attempts)
|
||||||
assertNull(repo.rows[1]!!.nextAttemptAt)
|
assertNull(repo.rows[1]!!.nextAttemptAt)
|
||||||
assertNull(repo.rows[1]!!.processingStartedAt)
|
|
||||||
assertEquals(ProcStatus.DEAD, repo.rows[2]!!.state) // MALFORMED 永不被重放
|
assertEquals(ProcStatus.DEAD, repo.rows[2]!!.state) // MALFORMED 永不被重放
|
||||||
assertEquals(ProcStatus.PENDING, repo.rows[3]!!.state)
|
assertEquals(ProcStatus.PENDING, repo.rows[3]!!.state)
|
||||||
}
|
}
|
||||||
|
|||||||
Reference in New Issue
Block a user