diff --git a/src/main/kotlin/com/gzzn/omms/msgexchange/domain/ProcState.kt b/src/main/kotlin/com/gzzn/omms/msgexchange/domain/ProcState.kt index 84ff96d..5d5f32d 100644 --- a/src/main/kotlin/com/gzzn/omms/msgexchange/domain/ProcState.kt +++ b/src/main/kotlin/com/gzzn/omms/msgexchange/domain/ProcState.kt @@ -86,12 +86,5 @@ data class ProcState( val backfillAbandonedAt: Instant? = null, /** 放弃原因（`MISSING_ROW` / `TRANSIENT_DEADLINE`），供人工对账与恢复判断。 */ val backfillAbandonedReason: String? = null, - /** - * 首次被主泵取得的时刻；重试不刷新。 - * - * 当前没有判据消费它（终态只看尝试上限），保留作排障与后续扩展；写入方是 - * [com.gzzn.omms.msgexchange.infra.persistence.ProcStateRepository.markProcessingStartedIfAbsent]。 - */ - val processingStartedAt: Instant? = null, val updatedAt: Instant = Instant.now(), ) diff --git a/src/main/kotlin/com/gzzn/omms/msgexchange/infra/persistence/Repositories.kt b/src/main/kotlin/com/gzzn/omms/msgexchange/infra/persistence/Repositories.kt index 2619745..1de4855 100644 --- a/src/main/kotlin/com/gzzn/omms/msgexchange/infra/persistence/Repositories.kt +++ b/src/main/kotlin/com/gzzn/omms/msgexchange/infra/persistence/Repositories.kt @@ -59,9 +59,6 @@ interface ProcStateRepository { /** 当前队头：还没处理完的消息里 ID 最小的那条。 */ fun headUnfinished(): ProcState? - /** 首次开始处理时记下稳定起点；重试不覆盖。 */ - fun markProcessingStartedIfAbsent(msgId: Long, now: Instant) - /** 给消息绑定业务身份；返回 false 表示这个身份已经被另一条消息占了（业务重复）。 */ fun tryBindIdentity(msgId: Long, identityKey: String): Boolean diff --git a/src/main/kotlin/com/gzzn/omms/msgexchange/infra/persistence/jdbc/JdbcPgRepositories.kt b/src/main/kotlin/com/gzzn/omms/msgexchange/infra/persistence/jdbc/JdbcPgRepositories.kt index 6319def..41d72ce 100644 --- a/src/main/kotlin/com/gzzn/omms/msgexchange/infra/persistence/jdbc/JdbcPgRepositories.kt +++ b/src/main/kotlin/com/gzzn/omms/msgexchange/infra/persistence/jdbc/JdbcPgRepositories.kt @@ -107,16 +107,6 @@ class JdbcProcStateRepository( ::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 { ownerOfIdentity(identityKey)?.let { owner -> return owner == msgId @@ -286,7 +276,7 @@ class JdbcProcStateRepository( if (errorClasses.isEmpty()) return 0 val placeholders = errorClasses.joinToString(",") { "?" } 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)", { ps -> ps.setTimestamp(1, clock.instant().toSqlTimestamp()) @@ -339,7 +329,6 @@ class JdbcProcStateRepository( backfillError = rs.getString("backfill_error"), backfillAbandonedAt = rs.getInstant("backfill_abandoned_at"), backfillAbandonedReason = rs.getString("backfill_abandoned_reason"), - processingStartedAt = rs.getInstant("processing_started_at"), updatedAt = rs.getInstant("updated_at") ?: clock.instant(), ) @@ -347,7 +336,7 @@ class JdbcProcStateRepository( const val SELECT_PROC = "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, " + - "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" } } diff --git a/src/main/kotlin/com/gzzn/omms/msgexchange/infra/stub/StubRepositories.kt b/src/main/kotlin/com/gzzn/omms/msgexchange/infra/stub/StubRepositories.kt index 3a84931..815118a 100644 --- a/src/main/kotlin/com/gzzn/omms/msgexchange/infra/stub/StubRepositories.kt +++ b/src/main/kotlin/com/gzzn/omms/msgexchange/infra/stub/StubRepositories.kt @@ -84,12 +84,6 @@ class StubProcState : ProcStateRepository { override fun headUnfinished(): ProcState? = 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 { val owner = bound[identityKey] if (owner != null && owner != msgId) return false @@ -216,7 +210,7 @@ class StubProcState : ProcStateRepository { var n = 0 rows.forEach { (id, s) -> 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++ } } diff --git a/src/main/kotlin/com/gzzn/omms/msgexchange/processing/Pump.kt b/src/main/kotlin/com/gzzn/omms/msgexchange/processing/Pump.kt index f6d5ccd..9f409ef 100644 --- a/src/main/kotlin/com/gzzn/omms/msgexchange/processing/Pump.kt +++ b/src/main/kotlin/com/gzzn/omms/msgexchange/processing/Pump.kt @@ -103,10 +103,7 @@ class Pump( head.state == ProcStatus.FAILED && (head.nextAttemptAt ?: Instant.EPOCH) > now -> sleepQuietly(Duration.between(now, head.nextAttemptAt)) // 其余情况（新消息，或退避到期的重试）交给处理入口 - else -> { - procState.markProcessingStartedIfAbsent(head.msgId, now) - processor.processOne(head) - } + else -> processor.processOne(head) } } diff --git a/src/main/resources/db/migration/V7__drop_processing_started_at.sql b/src/main/resources/db/migration/V7__drop_processing_started_at.sql new file mode 100644 index 0000000..991aec4 --- /dev/null +++ b/src/main/resources/db/migration/V7__drop_processing_started_at.sql @@ -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; diff --git a/src/test/kotlin/com/gzzn/omms/msgexchange/infra/persistence/jdbc/FlywayMigrationTest.kt b/src/test/kotlin/com/gzzn/omms/msgexchange/infra/persistence/jdbc/FlywayMigrationTest.kt index 7d67a35..cda37e8 100644 --- a/src/test/kotlin/com/gzzn/omms/msgexchange/infra/persistence/jdbc/FlywayMigrationTest.kt +++ b/src/test/kotlin/com/gzzn/omms/msgexchange/infra/persistence/jdbc/FlywayMigrationTest.kt @@ -48,7 +48,7 @@ class FlywayMigrationTest { while (rs.next()) { 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("V1__flight_state_baseline.sql", records[0].second) assertEquals("2", records[1].first) @@ -59,6 +59,10 @@ class FlywayMigrationTest { assertEquals("V4__backfill_closure.sql", records[3].second) assertEquals("5", records[4].first) 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 }) } @@ -88,7 +92,7 @@ class FlywayMigrationTest { "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', " + "'backfill_attempts', 'backfill_error', 'backfill_abandoned_at', " + - "'backfill_abandoned_reason', 'processing_started_at')", + "'backfill_abandoned_reason')", ).use { rs -> val cols = mutableSetOf() while (rs.next()) cols.add(rs.getString("column_name")) @@ -96,12 +100,20 @@ class FlywayMigrationTest { setOf( "received_at", "enqueued_at", "backfill_at", "backfill_next_at", "backfill_attempts", "backfill_error", "backfill_abandoned_at", "backfill_abandoned_reason", - "processing_started_at", ), 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） stmt.executeQuery( "SELECT is_nullable FROM information_schema.columns " + diff --git a/src/test/kotlin/com/gzzn/omms/msgexchange/infra/persistence/jdbc/InboxLifecycleJdbcSqlTest.kt b/src/test/kotlin/com/gzzn/omms/msgexchange/infra/persistence/jdbc/InboxLifecycleJdbcSqlTest.kt index afd663e..26ae4c2 100644 --- a/src/test/kotlin/com/gzzn/omms/msgexchange/infra/persistence/jdbc/InboxLifecycleJdbcSqlTest.kt +++ b/src/test/kotlin/com/gzzn/omms/msgexchange/infra/persistence/jdbc/InboxLifecycleJdbcSqlTest.kt @@ -142,15 +142,6 @@ class InboxLifecycleJdbcSqlTest { 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 fun `pipeline transaction rolls back terminal and outbox writes after a late failure`() { seed(11L, t0) @@ -232,7 +223,6 @@ class InboxLifecycleJdbcSqlTest { backfill_error VARCHAR(512), backfill_abandoned_at TIMESTAMP WITH TIME ZONE, backfill_abandoned_reason VARCHAR(64), - processing_started_at TIMESTAMP WITH TIME ZONE, updated_at TIMESTAMP WITH TIME ZONE NOT NULL, CONSTRAINT uk_proc_identity UNIQUE (identity_key) ) diff --git a/src/test/kotlin/com/gzzn/omms/msgexchange/infra/retry/ReplayServiceTest.kt b/src/test/kotlin/com/gzzn/omms/msgexchange/infra/retry/ReplayServiceTest.kt index 1db0766..9f0f42f 100644 --- a/src/test/kotlin/com/gzzn/omms/msgexchange/infra/retry/ReplayServiceTest.kt +++ b/src/test/kotlin/com/gzzn/omms/msgexchange/infra/retry/ReplayServiceTest.kt @@ -22,7 +22,6 @@ class ReplayServiceTest { fun seed(id: Long, status: ProcStatus, ec: ErrorClass?) { rows[id] = ProcState( 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 findSuccessTerminal(msgId: Long): Boolean = rows[msgId]?.state == ProcStatus.SUCCEEDED override fun headUnfinished(): ProcState? = null - override fun markProcessingStartedIfAbsent(msgId: Long, now: Instant) = Unit override fun tryBindIdentity(msgId: Long, identityKey: String): Boolean = true override fun ownerOfIdentity(identityKey: String): Long? = null override fun update( @@ -69,7 +67,7 @@ class ReplayServiceTest { val s = rows[id]!! if (s.errorClass != null && 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++ } } @@ -92,7 +90,6 @@ class ReplayServiceTest { assertEquals(ProcStatus.PENDING, repo.rows[1]!!.state) assertEquals(0, repo.rows[1]!!.attempts) assertNull(repo.rows[1]!!.nextAttemptAt) - assertNull(repo.rows[1]!!.processingStartedAt) assertEquals(ProcStatus.DEAD, repo.rows[2]!!.state) // MALFORMED 永不被重放 assertEquals(ProcStatus.PENDING, repo.rows[3]!!.state) }