diff --git a/src/main/kotlin/com/gzzn/omms/msgexchange/codec/ErorPayload.kt b/src/main/kotlin/com/gzzn/omms/msgexchange/codec/ErorPayload.kt new file mode 100644 index 0000000..7f33b43 --- /dev/null +++ b/src/main/kotlin/com/gzzn/omms/msgexchange/codec/ErorPayload.kt @@ -0,0 +1,9 @@ +package com.gzzn.omms.msgexchange.codec + +/** AODB EROR 回报体（SIS 3.44）：指向触发错误的源出站 META。 */ +data class ErorPayload( + val seqs: Long, + val typs: String, + val stys: String, + val etex: String, +) diff --git a/src/main/kotlin/com/gzzn/omms/msgexchange/codec/JacksonXmlCodec.kt b/src/main/kotlin/com/gzzn/omms/msgexchange/codec/JacksonXmlCodec.kt index 1fb9ddc..b18f411 100644 --- a/src/main/kotlin/com/gzzn/omms/msgexchange/codec/JacksonXmlCodec.kt +++ b/src/main/kotlin/com/gzzn/omms/msgexchange/codec/JacksonXmlCodec.kt @@ -9,6 +9,7 @@ import com.gzzn.omms.msgexchange.domain.DecodedMessage import com.gzzn.omms.msgexchange.domain.ErrorClass import com.gzzn.omms.msgexchange.domain.MetaFields import com.gzzn.omms.msgexchange.domain.MsgKind +import com.gzzn.omms.msgexchange.domain.OutboundRequestKeys import jakarta.inject.Singleton import javax.xml.stream.XMLInputFactory @@ -68,7 +69,14 @@ class JacksonXmlCodec : XmlCodec { MsgKind.Fdel -> msg.flop?.let(SisWireMapper::flopPayload) // 参考数据类别报文的正文结构随 13 类各异（ACM2-93 实装 ReferenceDataProcessor 时绑定） is MsgKind.RefData -> null - MsgKind.Eror -> null + MsgKind.Eror -> msg.eror?.let { + val seqs = it.seqs ?: return DecodeResult.Err(DecodeFailure(ErrorClass.MALFORMED, "missing-eror-seqs")) + val typs = it.typs?.trim()?.takeIf { t -> t.isNotEmpty() } + ?: return DecodeResult.Err(DecodeFailure(ErrorClass.MALFORMED, "missing-eror-typs")) + val stys = it.stys?.trim()?.takeIf { t -> t.isNotEmpty() } + ?: return DecodeResult.Err(DecodeFailure(ErrorClass.MALFORMED, "missing-eror-stys")) + ErorPayload(seqs, typs.uppercase(), stys.uppercase(), it.etex?.trim().orEmpty()) + } is MsgKind.Unsupported -> null } @@ -82,8 +90,36 @@ class JacksonXmlCodec : XmlCodec { ) } - override fun encodeRqrd(kind: String, rangeJson: String): String = - """RQFD""" + override fun encodeOutboundRqfd(seqn: Long, dttm: Long): String = + writeOutbound( + SisOutboundMessageXml( + meta = outboundMeta("RQFD", "NONE", seqn, dttm), + rqfd = RqfdXml(), + ), + ) + + override fun encodeOutboundRqrd(styp: String, seqn: Long, dttm: Long, rtyp: String?): String { + val body = if (rtyp != null) RqrdXml(rtyp = rtyp) else RqrdXml() + return writeOutbound( + SisOutboundMessageXml( + meta = outboundMeta("RQRD", styp.uppercase(), seqn, dttm), + rqrd = body, + ), + ) + } + + private fun outboundMeta(type: String, styp: String, seqn: Long, dttm: Long) = SisMetaXml( + sndr = OutboundRequestKeys.SENDER, + seqn = seqn, + dttm = dttm, + type = type, + styp = styp, + ) + + private fun writeOutbound(msg: SisOutboundMessageXml): String = + mapper.writeValueAsString(msg).let { xml -> + if (xml.startsWith("$xml""" + } /** * 子类型分派必须是**白名单**：未知的 SCHD 子类型不能静默当成全量日计划（DNLD）—— diff --git a/src/main/kotlin/com/gzzn/omms/msgexchange/codec/SisMessageBody.kt b/src/main/kotlin/com/gzzn/omms/msgexchange/codec/SisMessageBody.kt index 6b8a7fe..a000a87 100644 --- a/src/main/kotlin/com/gzzn/omms/msgexchange/codec/SisMessageBody.kt +++ b/src/main/kotlin/com/gzzn/omms/msgexchange/codec/SisMessageBody.kt @@ -56,6 +56,34 @@ data class SisMessageXml( @param:JacksonXmlProperty(localName = "META") val meta: SisMetaXml? = null, @param:JacksonXmlProperty(localName = "SCHD") val schd: SchdXml? = null, @param:JacksonXmlProperty(localName = "FLOP") val flop: FlightRecordXml? = null, + @param:JacksonXmlProperty(localName = "EROR") val eror: ErorXml? = null, + @param:JacksonXmlProperty(localName = "RQFD") val rqfd: RqfdXml? = null, + @param:JacksonXmlProperty(localName = "RQRD") val rqrd: RqrdXml? = null, +) + +@JsonIgnoreProperties(ignoreUnknown = true) +data class ErorXml( + @param:JacksonXmlProperty(localName = "SEQS") val seqs: Long? = null, + @param:JacksonXmlProperty(localName = "TYPS") val typs: String? = null, + @param:JacksonXmlProperty(localName = "STYS") val stys: String? = null, + @param:JacksonXmlProperty(localName = "ETEX") val etex: String? = null, +) + +/** 出站 RQFD 空体占位（C-4 全量不带 STDB/STDE）。 */ +@JsonIgnoreProperties(ignoreUnknown = true) +class RqfdXml + +@JsonIgnoreProperties(ignoreUnknown = true) +data class RqrdXml( + @param:JacksonXmlProperty(localName = "RTYP") val rtyp: String? = null, +) + +@JsonIgnoreProperties(ignoreUnknown = true) +@JacksonXmlRootElement(localName = "MSG") +data class SisOutboundMessageXml( + @param:JacksonXmlProperty(localName = "META") val meta: SisMetaXml, + @param:JacksonXmlProperty(localName = "RQFD") val rqfd: RqfdXml? = null, + @param:JacksonXmlProperty(localName = "RQRD") val rqrd: RqrdXml? = null, ) @JsonIgnoreProperties(ignoreUnknown = true) diff --git a/src/main/kotlin/com/gzzn/omms/msgexchange/codec/XmlCodec.kt b/src/main/kotlin/com/gzzn/omms/msgexchange/codec/XmlCodec.kt index 156ecee..e55fc9a 100644 --- a/src/main/kotlin/com/gzzn/omms/msgexchange/codec/XmlCodec.kt +++ b/src/main/kotlin/com/gzzn/omms/msgexchange/codec/XmlCodec.kt @@ -21,5 +21,10 @@ sealed interface DecodeResult { */ interface XmlCodec { fun decode(rawXml: String): DecodeResult - fun encodeRqrd(kind: String, rangeJson: String): String + + /** C-4：RQFD STYP=NONE，空 RQFD 体。 */ + fun encodeOutboundRqfd(seqn: Long, dttm: Long): String + + /** C-4：RQRD，STYP 为参考数据子类型；RSTA 时可带 RTYP。 */ + fun encodeOutboundRqrd(styp: String, seqn: Long, dttm: Long, rtyp: String? = null): String } diff --git a/src/main/kotlin/com/gzzn/omms/msgexchange/config/PipelineProps.kt b/src/main/kotlin/com/gzzn/omms/msgexchange/config/PipelineProps.kt index 556829e..ab7a5a8 100644 --- a/src/main/kotlin/com/gzzn/omms/msgexchange/config/PipelineProps.kt +++ b/src/main/kotlin/com/gzzn/omms/msgexchange/config/PipelineProps.kt @@ -15,6 +15,7 @@ class PipelineProps { var registerEureka: Boolean = true var pipeline: Pipeline = Pipeline() var schd: Schd = Schd() + var outbound: Outbound = Outbound() var identity: Identity = Identity() @ConfigurationProperties("pipeline") @@ -115,6 +116,15 @@ class PipelineProps { var snapshotBatch: Int = 200 } + @ConfigurationProperties("outbound") + class Outbound { + /** 出站请求落信后等待应答的最长时限（`US-09`；具体取值待 Q）。 */ + var responseTimeout: Duration = Duration.ofMinutes(30) + + var routingRqfd: String = "OMMSRQFD" + var routingRqrd: String = "OMMSRQRD" + } + @ConfigurationProperties("identity") class Identity { /** CONFIRM（矩阵 #11）：SEQN 重置作用域确认前保持 false，计算集中此处（I3）。 */ diff --git a/src/main/kotlin/com/gzzn/omms/msgexchange/domain/OutboundRequestKeys.kt b/src/main/kotlin/com/gzzn/omms/msgexchange/domain/OutboundRequestKeys.kt new file mode 100644 index 0000000..b1fba6f --- /dev/null +++ b/src/main/kotlin/com/gzzn/omms/msgexchange/domain/OutboundRequestKeys.kt @@ -0,0 +1,15 @@ +package com.gzzn.omms.msgexchange.domain + +/** 出站请求在 REQ_TRACK 与 C-4 编码中的稳定键。 */ +object OutboundRequestKeys { + const val SENDER = "OMMS" + const val RQFD_REQ_TYPE = "RQFD-NONE" + const val RQRD_REQ_TYPE = "RQRD" + + /** EROR 体 TYPS/STYS → REQ_TRACK.req_type；无法配对时 null。 */ + fun reqTypeFromEror(typs: String, stys: String): String? = when (typs.uppercase()) { + "RQFD" -> if (stys.uppercase() == "NONE") RQFD_REQ_TYPE else null + "RQRD" -> RQRD_REQ_TYPE + else -> null + } +} 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 399c094..f27f0d4 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 @@ -306,7 +306,7 @@ interface SnapshotLogRepository { * 同一类请求同时只保留一条有效，发新请求时把旧的置为已过期。 */ interface ReqTrackRepository { - enum class ReqState { PENDING, SENT, DONE, EXPIRED } + enum class ReqState { PENDING, SENT, DONE, EXPIRED, FAILED } data class Req( val reqId: Long, @@ -315,7 +315,10 @@ interface ReqTrackRepository { val sender: String, val state: ReqState, val coutmsgsId: Long? = null, + val outboundSeqn: Long? = null, + val writeUncertain: Boolean = false, val sentAt: Instant? = null, + val createdAt: Instant? = null, ) /** 登记新请求：同类（类型+运营日+发送方）旧有效请求先置 EXPIRED。 */ @@ -329,11 +332,35 @@ interface ReqTrackRepository { */ fun completeLatest(reqType: String, operationDay: LocalDate, sender: String): Boolean - fun linkCoutmsgs(reqId: Long, coutmsgsId: Long) + fun linkCoutmsgs(reqId: Long, coutmsgsId: Long, outboundSeqn: Long) fun markSent(reqId: Long, sentAt: Instant) fun expire(reqId: Long) + + fun markFailed(reqId: Long) + + fun markWriteUncertain(reqId: Long) + + /** 待落信：PENDING 且未标记 write_uncertain。 */ + fun listPendingDispatch(limit: Int): List + + /** 已落信等待应答且超过时限：标记 EXPIRED 并返回条数。 */ + fun expireSentOlderThan(cutoff: Instant): Int + + /** EROR：按出站 SEQN + 请求族匹配开放 SENT。 */ + fun findOpenSentBySeqn(reqType: String, operationDay: LocalDate, sender: String, outboundSeqn: Long): Req? +} + +/** 共享 MySQL COUTMSGS 出站信箱（只写、不改表）。 */ +interface CoutmsgOutboxRepository { + enum class InsertOutcome { CONFIRMED, AMBIGUOUS } + + data class InsertResult(val outcome: InsertOutcome, val coutmsgsId: Long?) + + fun nextOutboundSeqn(): Long + + fun insertMessage(rawXml: String, routingId: String): InsertResult } /** diff --git a/src/main/kotlin/com/gzzn/omms/msgexchange/infra/persistence/jdbc/JdbcCoutmsgOutboxRepository.kt b/src/main/kotlin/com/gzzn/omms/msgexchange/infra/persistence/jdbc/JdbcCoutmsgOutboxRepository.kt new file mode 100644 index 0000000..2c57405 --- /dev/null +++ b/src/main/kotlin/com/gzzn/omms/msgexchange/infra/persistence/jdbc/JdbcCoutmsgOutboxRepository.kt @@ -0,0 +1,42 @@ +package com.gzzn.omms.msgexchange.infra.persistence.jdbc + +import com.gzzn.omms.msgexchange.infra.persistence.CoutmsgOutboxRepository +import io.micronaut.context.annotation.Requires +import jakarta.inject.Named +import jakarta.inject.Singleton +import javax.sql.DataSource + +@Singleton +@Requires(property = "msgx.stubs", notEquals = "true") +@Requires(property = "mailbox.shared-mysql.enabled", value = "true") +class JdbcCoutmsgOutboxRepository( + @Named("mailbox") private val ds: DataSource, +) : CoutmsgOutboxRepository { + + override fun nextOutboundSeqn(): Long = + ds.queryOne("SELECT COALESCE(MAX(COUTMSGS_ID), 0) + 1 FROM coutmsgs", { _ -> }) { rs -> + rs.getLong(1) + } ?: 1L + + override fun insertMessage(rawXml: String, routingId: String): CoutmsgOutboxRepository.InsertResult = + try { + val id = ds.updateReturningLong( + """ + INSERT INTO coutmsgs (COUTMSGS_CLOB_MSG, COUTMSGS_DATE_INSERTED, ROUTINGID) + VALUES (?, CURRENT_TIMESTAMP, ?) + """.trimIndent(), + ) { ps -> + ps.setString(1, rawXml) + ps.setString(2, routingId) + } + CoutmsgOutboxRepository.InsertResult(CoutmsgOutboxRepository.InsertOutcome.CONFIRMED, id) + } catch (e: IllegalStateException) { + CoutmsgOutboxRepository.InsertResult(CoutmsgOutboxRepository.InsertOutcome.AMBIGUOUS, null) + } catch (e: Exception) { + if (e.cause is IllegalStateException) { + CoutmsgOutboxRepository.InsertResult(CoutmsgOutboxRepository.InsertOutcome.AMBIGUOUS, null) + } else { + throw e + } + } +} 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 eb5b017..29e91cd 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 @@ -961,8 +961,18 @@ class JdbcReqTrackRepository( private val ds: DataSource, private val clock: Clock, ) : ReqTrackRepository { - override fun insert(reqType: String, operationDay: LocalDate, sender: String): Long = - ds.updateReturningLong( + override fun insert(reqType: String, operationDay: LocalDate, sender: String): Long { + ds.update( + """ + UPDATE req_track SET state = 'EXPIRED' + WHERE req_type = ? AND operation_day = ? AND sender = ? AND state IN ('PENDING', 'SENT') + """.trimIndent(), + ) { ps -> + ps.setString(1, reqType) + ps.setDate(2, java.sql.Date.valueOf(operationDay)) + ps.setString(3, sender) + } + return ds.updateReturningLong( "INSERT INTO req_track (req_type, operation_day, sender, state, created_at) VALUES (?, ?, ?, 'PENDING', ?) RETURNING req_id", { ps -> ps.setString(1, reqType) @@ -971,6 +981,7 @@ class JdbcReqTrackRepository( ps.setTimestamp(4, clock.instant().toSqlTimestamp()) }, ) + } override fun findLatest( reqType: String, @@ -1015,8 +1026,15 @@ class JdbcReqTrackRepository( }, ) == 1 - override fun linkCoutmsgs(reqId: Long, coutmsgsId: Long) { - ds.update("UPDATE req_track SET coutmsgs_id = ? WHERE req_id = ?", { ps -> ps.setLong(1, coutmsgsId); ps.setLong(2, reqId) }) + override fun linkCoutmsgs(reqId: Long, coutmsgsId: Long, outboundSeqn: Long) { + ds.update( + "UPDATE req_track SET coutmsgs_id = ?, outbound_seqn = ? WHERE req_id = ?", + { ps -> + ps.setLong(1, coutmsgsId) + ps.setLong(2, outboundSeqn) + ps.setLong(3, reqId) + }, + ) } override fun markSent(reqId: Long, sentAt: Instant) { @@ -1030,6 +1048,62 @@ class JdbcReqTrackRepository( ds.update("UPDATE req_track SET state = 'EXPIRED' WHERE req_id = ?", { ps -> ps.setLong(1, reqId) }) } + override fun markFailed(reqId: Long) { + ds.update( + "UPDATE req_track SET state = 'FAILED', completed_at = ? WHERE req_id = ?", + { ps -> + ps.setTimestamp(1, clock.instant().toSqlTimestamp()) + ps.setLong(2, reqId) + }, + ) + } + + override fun markWriteUncertain(reqId: Long) { + ds.update("UPDATE req_track SET write_uncertain = TRUE WHERE req_id = ?", { ps -> ps.setLong(1, reqId) }) + } + + override fun listPendingDispatch(limit: Int): List = + ds.query( + """ + SELECT * FROM req_track + WHERE state = 'PENDING' AND write_uncertain = FALSE + ORDER BY created_at ASC, req_id ASC + LIMIT ? + """.trimIndent(), + { ps -> ps.setInt(1, limit) }, + ::mapReq, + ) + + override fun expireSentOlderThan(cutoff: Instant): Int = + ds.update( + """ + UPDATE req_track SET state = 'EXPIRED' + WHERE state = 'SENT' AND sent_at IS NOT NULL AND sent_at < ? + """.trimIndent(), + ) { ps -> ps.setTimestamp(1, cutoff.toSqlTimestamp()) } + + override fun findOpenSentBySeqn( + reqType: String, + operationDay: LocalDate, + sender: String, + outboundSeqn: Long, + ): ReqTrackRepository.Req? = + ds.queryOne( + """ + SELECT * FROM req_track + WHERE req_type = ? AND operation_day = ? AND sender = ? + AND state = 'SENT' AND outbound_seqn = ? + ORDER BY created_at DESC, req_id DESC LIMIT 1 + """.trimIndent(), + { ps -> + ps.setString(1, reqType) + ps.setDate(2, java.sql.Date.valueOf(operationDay)) + ps.setString(3, sender) + ps.setLong(4, outboundSeqn) + }, + ::mapReq, + ) + private fun mapReq(rs: ResultSet) = ReqTrackRepository.Req( reqId = rs.getLong("req_id"), reqType = rs.getString("req_type"), @@ -1037,6 +1111,9 @@ class JdbcReqTrackRepository( sender = rs.getString("sender"), state = ReqTrackRepository.ReqState.valueOf(rs.getString("state")), coutmsgsId = rs.getLong("coutmsgs_id").takeIf { !rs.wasNull() }, + outboundSeqn = rs.getLong("outbound_seqn").takeIf { !rs.wasNull() }, + writeUncertain = rs.getBoolean("write_uncertain"), sentAt = rs.getInstant("sent_at"), + createdAt = rs.getInstant("created_at"), ) } 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 a9b7dfd..b211210 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 @@ -24,6 +24,7 @@ import com.gzzn.omms.msgexchange.infra.persistence.PersistOutcome import com.gzzn.omms.msgexchange.infra.persistence.PipelineLockRepository import com.gzzn.omms.msgexchange.infra.persistence.PipelineTransactionManager import com.gzzn.omms.msgexchange.infra.persistence.ProcStateRepository +import com.gzzn.omms.msgexchange.infra.persistence.CoutmsgOutboxRepository import com.gzzn.omms.msgexchange.infra.persistence.ReqTrackRepository import com.gzzn.omms.msgexchange.domain.SnapshotLogEntry import com.gzzn.omms.msgexchange.infra.persistence.SnapshotLogRepository @@ -464,7 +465,7 @@ class StubSnapshotLog : SnapshotLogRepository, SnapshotLogPurge { @Singleton @Requires(property = "msgx.stubs", value = "true") -class StubReqTrack : ReqTrackRepository { +class StubReqTrack(private val clock: Clock = Clock.systemUTC()) : ReqTrackRepository { val rows = linkedMapOf() private val ids = AtomicLong(0) @@ -477,7 +478,10 @@ class StubReqTrack : ReqTrackRepository { (it.state == ReqTrackRepository.ReqState.PENDING || it.state == ReqTrackRepository.ReqState.SENT) }.forEach { rows[it.reqId] = it.copy(state = ReqTrackRepository.ReqState.EXPIRED) } val id = ids.incrementAndGet() - rows[id] = ReqTrackRepository.Req(id, reqType, operationDay, sender, ReqTrackRepository.ReqState.PENDING) + rows[id] = ReqTrackRepository.Req( + id, reqType, operationDay, sender, ReqTrackRepository.ReqState.PENDING, + createdAt = clock.instant(), + ) return id } @@ -498,8 +502,8 @@ class StubReqTrack : ReqTrackRepository { return true } - override fun linkCoutmsgs(reqId: Long, coutmsgsId: Long) { - rows[reqId]?.let { rows[reqId] = it.copy(coutmsgsId = coutmsgsId) } + override fun linkCoutmsgs(reqId: Long, coutmsgsId: Long, outboundSeqn: Long) { + rows[reqId]?.let { rows[reqId] = it.copy(coutmsgsId = coutmsgsId, outboundSeqn = outboundSeqn) } } override fun markSent(reqId: Long, sentAt: Instant) { @@ -509,6 +513,65 @@ class StubReqTrack : ReqTrackRepository { override fun expire(reqId: Long) { rows[reqId]?.let { rows[reqId] = it.copy(state = ReqTrackRepository.ReqState.EXPIRED) } } + + override fun markFailed(reqId: Long) { + rows[reqId]?.let { rows[reqId] = it.copy(state = ReqTrackRepository.ReqState.FAILED) } + } + + override fun markWriteUncertain(reqId: Long) { + rows[reqId]?.let { rows[reqId] = it.copy(writeUncertain = true) } + } + + override fun listPendingDispatch(limit: Int): List = + rows.values.filter { it.state == ReqTrackRepository.ReqState.PENDING && !it.writeUncertain } + .sortedWith(compareBy({ it.createdAt }, { it.reqId })) + .take(limit) + + override fun expireSentOlderThan(cutoff: Instant): Int { + var n = 0 + rows.values.filter { it.state == ReqTrackRepository.ReqState.SENT && it.sentAt != null && it.sentAt!! < cutoff } + .forEach { + rows[it.reqId] = it.copy(state = ReqTrackRepository.ReqState.EXPIRED) + n++ + } + return n + } + + override fun findOpenSentBySeqn( + reqType: String, + operationDay: LocalDate, + sender: String, + outboundSeqn: Long, + ): ReqTrackRepository.Req? = + rows.values.filter { + it.reqType == reqType && it.operationDay == operationDay && it.sender == sender && + it.state == ReqTrackRepository.ReqState.SENT && it.outboundSeqn == outboundSeqn + }.maxByOrNull { it.reqId } +} + +@Singleton +@Requires(property = "msgx.stubs", value = "true") +class StubCoutmsgOutbox : CoutmsgOutboxRepository { + val messages = linkedMapOf() + private val ids = AtomicLong(0) + + fun clear() { + messages.clear() + ids.set(0) + } + + var nextInsertOutcome: CoutmsgOutboxRepository.InsertOutcome = CoutmsgOutboxRepository.InsertOutcome.CONFIRMED + + override fun nextOutboundSeqn(): Long = (messages.keys.maxOrNull() ?: 0L) + 1L + + override fun insertMessage(rawXml: String, routingId: String): CoutmsgOutboxRepository.InsertResult { + if (nextInsertOutcome == CoutmsgOutboxRepository.InsertOutcome.AMBIGUOUS) { + return CoutmsgOutboxRepository.InsertResult(CoutmsgOutboxRepository.InsertOutcome.AMBIGUOUS, null) + } + val id = ids.incrementAndGet() + messages[id] = rawXml + return CoutmsgOutboxRepository.InsertResult(CoutmsgOutboxRepository.InsertOutcome.CONFIRMED, id) + } } /** 内存版共享信箱：可以模拟上游写入、库方清除，并记录哪些行被打上了处理标记。 */ diff --git a/src/main/kotlin/com/gzzn/omms/msgexchange/ingress/InboxController.kt b/src/main/kotlin/com/gzzn/omms/msgexchange/ingress/InboxController.kt index 8fc7588..3bb0e88 100644 --- a/src/main/kotlin/com/gzzn/omms/msgexchange/ingress/InboxController.kt +++ b/src/main/kotlin/com/gzzn/omms/msgexchange/ingress/InboxController.kt @@ -13,7 +13,10 @@ import io.micronaut.http.annotation.Produces * 生产上收报走的是轮询共享信箱表 CMINMSGS（InboxPoller），不是这个 Controller。 */ @Controller -class InboxController(private val inbox: InboxService) { +class InboxController( + private val inbox: InboxService, + private val schdSync: SchdSyncService, +) { @Post("/cminmsgs/send") @Produces(MediaType.TEXT_PLAIN) @@ -22,5 +25,14 @@ class InboxController(private val inbox: InboxService) { return HttpResponse.ok(receipt.msgId.toString()) // TODO: 先返回消息 ID 文本，等跟现役响应体逐字对拍过再定稿 } - // TODO(阶段2): 按契约清单补齐 /schd/sync、/kafka/topics/{name}/msgs 这两个接口。 + /** C-8：登记 RQFD 出站请求；开放槽占用时 409，不重复登记。 */ + @Post("/schd/sync") + @Produces(MediaType.TEXT_PLAIN) + fun schdSync(): HttpResponse = + when (val outcome = schdSync.trigger()) { + is SchdSyncService.Outcome.Registered -> + HttpResponse.ok(outcome.reqId.toString()) + SchdSyncService.Outcome.OpenExists -> + HttpResponse.status(io.micronaut.http.HttpStatus.CONFLICT).body("open-request-exists") + } } diff --git a/src/main/kotlin/com/gzzn/omms/msgexchange/ingress/SchdSyncService.kt b/src/main/kotlin/com/gzzn/omms/msgexchange/ingress/SchdSyncService.kt new file mode 100644 index 0000000..e323c0d --- /dev/null +++ b/src/main/kotlin/com/gzzn/omms/msgexchange/ingress/SchdSyncService.kt @@ -0,0 +1,18 @@ +package com.gzzn.omms.msgexchange.ingress + +import com.gzzn.omms.msgexchange.processing.OutboundRequestService +import jakarta.inject.Singleton + +@Singleton +class SchdSyncService(private val outbound: OutboundRequestService) { + sealed interface Outcome { + data class Registered(val reqId: Long) : Outcome + data object OpenExists : Outcome + } + + fun trigger(): Outcome = + when (val r = outbound.registerRqfdSync()) { + is OutboundRequestService.RegisterOutcome.Registered -> Outcome.Registered(r.reqId) + OutboundRequestService.RegisterOutcome.OpenExists -> Outcome.OpenExists + } +} diff --git a/src/main/kotlin/com/gzzn/omms/msgexchange/jobs/JobRunner.kt b/src/main/kotlin/com/gzzn/omms/msgexchange/jobs/JobRunner.kt index 181f4ca..df42681 100644 --- a/src/main/kotlin/com/gzzn/omms/msgexchange/jobs/JobRunner.kt +++ b/src/main/kotlin/com/gzzn/omms/msgexchange/jobs/JobRunner.kt @@ -26,6 +26,7 @@ class JobRunner( private val historySweep: HistorySweepJob, private val eventCleanup: EventCleanupJob, private val procStateCleanup: ProcStateCleanupJob, + private val reqTrackMaintenance: ReqTrackMaintenanceJob, private val clock: Clock, private val activity: JobActivity, private val props: PipelineProps, @@ -87,6 +88,10 @@ class JobRunner( if (procCleanup.deleted > 0) { log.info("proc_state cleanup: deleted {} expired terminal rows", procCleanup.deleted) } + val reqTrackOutcome = reqTrackMaintenance.run() + if (reqTrackOutcome.dispatched > 0 || reqTrackOutcome.expired > 0) { + log.info("req_track maintenance: dispatched={} expired={}", reqTrackOutcome.dispatched, reqTrackOutcome.expired) + } activity.tickFinished(startedAt, (System.nanoTime() - startedNanos) / 1_000_000, selected) } catch (e: InterruptedException) { throw e diff --git a/src/main/kotlin/com/gzzn/omms/msgexchange/jobs/ReqTrackMaintenanceJob.kt b/src/main/kotlin/com/gzzn/omms/msgexchange/jobs/ReqTrackMaintenanceJob.kt new file mode 100644 index 0000000..b9d625e --- /dev/null +++ b/src/main/kotlin/com/gzzn/omms/msgexchange/jobs/ReqTrackMaintenanceJob.kt @@ -0,0 +1,15 @@ +package com.gzzn.omms.msgexchange.jobs + +import com.gzzn.omms.msgexchange.processing.OutboundRequestService +import jakarta.inject.Singleton + +/** 出站请求：补偿落信与应答超时（`G-REQ-TRACK`）。 */ +@Singleton +class ReqTrackMaintenanceJob(private val outbound: OutboundRequestService) { + data class Outcome(val dispatched: Int, val expired: Int) + + fun run(): Outcome = Outcome( + dispatched = outbound.dispatchPending(limit = 20), + expired = outbound.expireTimedOut(), + ) +} diff --git a/src/main/kotlin/com/gzzn/omms/msgexchange/processing/OutboundRequestService.kt b/src/main/kotlin/com/gzzn/omms/msgexchange/processing/OutboundRequestService.kt new file mode 100644 index 0000000..3a246ec --- /dev/null +++ b/src/main/kotlin/com/gzzn/omms/msgexchange/processing/OutboundRequestService.kt @@ -0,0 +1,137 @@ +package com.gzzn.omms.msgexchange.processing + +import com.gzzn.omms.msgexchange.codec.XmlCodec +import com.gzzn.omms.msgexchange.config.OperationDayProps +import com.gzzn.omms.msgexchange.config.PipelineProps +import com.gzzn.omms.msgexchange.domain.OutboundRequestKeys +import com.gzzn.omms.msgexchange.infra.persistence.CoutmsgOutboxRepository +import com.gzzn.omms.msgexchange.infra.persistence.ReqTrackRepository +import jakarta.inject.Singleton +import org.slf4j.LoggerFactory +import java.time.Clock +import java.time.LocalDate +import java.time.ZoneId +import java.time.format.DateTimeFormatter + +/** + * 出站 REQ_TRACK 协调：登记、COUTMSGS 落信、超时与 EROR 失败（`G-REQ-TRACK`、`C-4`）。 + */ +@Singleton +class OutboundRequestService( + private val reqTrack: ReqTrackRepository, + private val outbox: CoutmsgOutboxRepository, + private val codec: XmlCodec, + private val props: PipelineProps, + operationDayProps: OperationDayProps, + private val clock: Clock, +) { + private val log = LoggerFactory.getLogger(OutboundRequestService::class.java) + private val zone: ZoneId = operationDayProps.zoneId() + private val cutoffHour = operationDayProps.cutoffHour.coerceIn(0, 23) + private val dttmFmt = DateTimeFormatter.ofPattern("yyyyMMddHHmmss") + + sealed interface RegisterOutcome { + data class Registered(val reqId: Long) : RegisterOutcome + data object OpenExists : RegisterOutcome + } + + fun currentOperationDay(): LocalDate { + val airport = clock.withZone(zone) + val now = java.time.LocalDateTime.now(airport) + val day = now.toLocalDate() + return if (now.hour < cutoffHour) day.minusDays(1) else day + } + + fun hasOpenRequest(reqType: String, operationDay: LocalDate = currentOperationDay()): Boolean = + reqTrack.findLatest( + reqType, + operationDay, + OutboundRequestKeys.SENDER, + listOf(ReqTrackRepository.ReqState.PENDING, ReqTrackRepository.ReqState.SENT), + ) != null + + fun registerRqfdSync(): RegisterOutcome { + val day = currentOperationDay() + if (hasOpenRequest(OutboundRequestKeys.RQFD_REQ_TYPE, day)) return RegisterOutcome.OpenExists + val reqId = reqTrack.insert(OutboundRequestKeys.RQFD_REQ_TYPE, day, OutboundRequestKeys.SENDER) + dispatchPending(limit = 1) + return RegisterOutcome.Registered(reqId) + } + + fun hasOpenSentRqfd(operationDay: LocalDate = currentOperationDay()): Boolean = + reqTrack.findLatest( + OutboundRequestKeys.RQFD_REQ_TYPE, + operationDay, + OutboundRequestKeys.SENDER, + listOf(ReqTrackRepository.ReqState.SENT), + ) != null + + fun completeRqfdResponse(operationDay: LocalDate = currentOperationDay()): Boolean = + reqTrack.completeLatest(OutboundRequestKeys.RQFD_REQ_TYPE, operationDay, OutboundRequestKeys.SENDER) + + fun dispatchPending(limit: Int = 10): Int { + var sent = 0 + reqTrack.listPendingDispatch(limit).forEach { req -> + if (dispatchOne(req)) sent++ + } + return sent + } + + fun expireTimedOut(): Int { + val cutoff = clock.instant().minus(props.outbound.responseTimeout) + return reqTrack.expireSentOlderThan(cutoff) + } + + fun failFromEror(seqs: Long, typs: String, stys: String): Boolean { + val reqType = OutboundRequestKeys.reqTypeFromEror(typs, stys) ?: return false + val day = currentOperationDay() + val open = reqTrack.findOpenSentBySeqn(reqType, day, OutboundRequestKeys.SENDER, seqs) ?: return false + reqTrack.markFailed(open.reqId) + log.warn( + "outbound request FAILED from EROR reqId={} type={} seqn={} typs={} stys={}", + open.reqId, reqType, seqs, typs, stys, + ) + return true + } + + private fun dispatchOne(req: ReqTrackRepository.Req): Boolean { + if (req.state != ReqTrackRepository.ReqState.PENDING || req.writeUncertain) return false + val seqn = outbox.nextOutboundSeqn() + val dttm = beijingMetaDttm() + val xml = when (req.reqType) { + OutboundRequestKeys.RQFD_REQ_TYPE -> codec.encodeOutboundRqfd(seqn, dttm) + OutboundRequestKeys.RQRD_REQ_TYPE -> codec.encodeOutboundRqrd("AIRL", seqn, dttm) + else -> { + log.warn("unknown req_type for dispatch reqId={} type={}", req.reqId, req.reqType) + return false + } + } + val routing = when (req.reqType) { + OutboundRequestKeys.RQFD_REQ_TYPE -> props.outbound.routingRqfd + else -> props.outbound.routingRqrd + } + val inserted = outbox.insertMessage(xml, routing) + return when (inserted.outcome) { + CoutmsgOutboxRepository.InsertOutcome.CONFIRMED -> { + val id = inserted.coutmsgsId ?: return false + reqTrack.linkCoutmsgs(req.reqId, id, seqn) + reqTrack.markSent(req.reqId, clock.instant()) + log.info("outbound COUTMSGS SENT reqId={} coutmsgsId={} seqn={}", req.reqId, id, seqn) + true + } + CoutmsgOutboxRepository.InsertOutcome.AMBIGUOUS -> { + reqTrack.markWriteUncertain(req.reqId) + log.error( + "outbound COUTMSGS write AMBIGUOUS reqId={} seqn={} — recorded, no blind resend", + req.reqId, seqn, + ) + false + } + } + } + + private fun beijingMetaDttm(): Long { + val ldt = java.time.LocalDateTime.ofInstant(clock.instant(), zone) + return dttmFmt.format(ldt).toLong() + } +} 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 6de640b..9a45f0d 100644 --- a/src/main/kotlin/com/gzzn/omms/msgexchange/processing/Pump.kt +++ b/src/main/kotlin/com/gzzn/omms/msgexchange/processing/Pump.kt @@ -1,5 +1,6 @@ package com.gzzn.omms.msgexchange.processing +import com.gzzn.omms.msgexchange.codec.ErorPayload import com.gzzn.omms.msgexchange.codec.FlopPayload import com.gzzn.omms.msgexchange.codec.ScheduleBody import com.gzzn.omms.msgexchange.codec.XmlCodec @@ -123,6 +124,7 @@ class MessageProcessor( private val flopProcessor: FlopProcessor, private val fdelProcessor: FdelProcessor, private val adftProcessor: AdftProcessor, + private val outbound: OutboundRequestService, private val procFailure: ProcFailure, private val props: PipelineProps, private val clock: Clock, @@ -201,8 +203,24 @@ class MessageProcessor( val body = decoded.body as? ScheduleBody if (body == null) return deadMalformed(head, "missing-schd-body") when (kind.subtype) { - MsgKind.SchdSubtype.DNLD, MsgKind.SchdSubtype.RESP -> + MsgKind.SchdSubtype.DNLD -> scheduleProcessor.applyScheduleRecords(head, decoded) + MsgKind.SchdSubtype.RESP -> { + if (!outbound.hasOpenSentRqfd()) { + log.info("SCHD-RESP without open RQFD -> SKIPPED msgId={} [G-RESP-GUARD]", head.msgId) + procState.markTerminal( + head.msgId, ProcStatus.SKIPPED, + lastError = "resp-guard:no-open-req", + now = clock.instant(), + ) + return + } + val respResult = scheduleProcessor.applyScheduleRecords(head, decoded) + if (respResult is ApplyResult.Succeeded || respResult is ApplyResult.ReplaySkipped) { + outbound.completeRqfdResponse() + } + respResult + } MsgKind.SchdSubtype.ADFT -> { val record = body.records.singleOrNull() if (record == null) return deadMalformed(head, "adft-needs-single-fltr") @@ -227,9 +245,20 @@ class MessageProcessor( return procFailure.fail(head, ErrorClass.UNSUPPORTED, "refdata-pending:${kind.type}") } MsgKind.Eror -> { - // US-09 AC3：须匹配出站请求；REQ_TRACK 协调器落地前（ACM2-92）先按可重试失败留队。 - log.warn("eror without handler -> FAILED(UNSUPPORTED) msgId={}", head.msgId) - return procFailure.fail(head, ErrorClass.UNSUPPORTED, "eror-pending") + val payload = decoded.body as? ErorPayload + ?: return deadMalformed(head, "missing-eror-body") + val matched = outbound.failFromEror(payload.seqs, payload.typs, payload.stys) + if (!matched) { + log.info("EROR without matching open outbound -> SKIPPED msgId={}", head.msgId) + procState.markTerminal( + head.msgId, ProcStatus.SKIPPED, + lastError = "eror:no-matching-req", + now = clock.instant(), + ) + } else { + procState.markTerminal(head.msgId, ProcStatus.SUCCEEDED, now = clock.instant()) + } + return } is MsgKind.Unsupported -> { // 合法但不支持的类型：跳过留档记 SKIPPED（US-03 AC2；markTerminal 同一条 UPDATE diff --git a/src/main/resources/db/migration/V4__req_track_outbound_seqn.sql b/src/main/resources/db/migration/V4__req_track_outbound_seqn.sql new file mode 100644 index 0000000..2fce20a --- /dev/null +++ b/src/main/resources/db/migration/V4__req_track_outbound_seqn.sql @@ -0,0 +1,3 @@ +-- REQ_TRACK：出站 SEQN（EROR 配对）与落信结果不明标记（不盲重发）。 +ALTER TABLE req_track ADD COLUMN IF NOT EXISTS outbound_seqn BIGINT; +ALTER TABLE req_track ADD COLUMN IF NOT EXISTS write_uncertain BOOLEAN NOT NULL DEFAULT FALSE; diff --git a/src/test/kotlin/com/gzzn/omms/msgexchange/codec/JacksonXmlCodecTest.kt b/src/test/kotlin/com/gzzn/omms/msgexchange/codec/JacksonXmlCodecTest.kt index 9a8b54a..c6fe0f5 100644 --- a/src/test/kotlin/com/gzzn/omms/msgexchange/codec/JacksonXmlCodecTest.kt +++ b/src/test/kotlin/com/gzzn/omms/msgexchange/codec/JacksonXmlCodecTest.kt @@ -16,6 +16,22 @@ class JacksonXmlCodecTest { private val codec = JacksonXmlCodec() + @Test + fun `encode outbound RQFD uses OMMS meta and empty body per C-4`() { + val xml = codec.encodeOutboundRqfd(1243L, 20021010090311L) + assertTrue(xml.contains("OMMS")) + assertTrue(xml.contains("RQFD")) + assertTrue(xml.contains("NONE")) + assertTrue(xml.contains("RQRD")) + assertTrue(xml.contains("AIRL")) + } + @Test fun `decode AODBGTDT sample extracts META and three gates`() { val raw = """ diff --git a/src/test/kotlin/com/gzzn/omms/msgexchange/jobs/JobRunnerTest.kt b/src/test/kotlin/com/gzzn/omms/msgexchange/jobs/JobRunnerTest.kt index 57dabca..7d4f9d0 100644 --- a/src/test/kotlin/com/gzzn/omms/msgexchange/jobs/JobRunnerTest.kt +++ b/src/test/kotlin/com/gzzn/omms/msgexchange/jobs/JobRunnerTest.kt @@ -14,8 +14,12 @@ import com.gzzn.omms.msgexchange.infra.stub.StubInbox import com.gzzn.omms.msgexchange.infra.stub.StubMsgEvents import com.gzzn.omms.msgexchange.infra.stub.StubPipelineLock import com.gzzn.omms.msgexchange.infra.stub.StubPipelineTx +import com.gzzn.omms.msgexchange.infra.stub.StubCoutmsgOutbox import com.gzzn.omms.msgexchange.infra.stub.StubProcState +import com.gzzn.omms.msgexchange.infra.stub.StubReqTrack +import com.gzzn.omms.msgexchange.codec.JacksonXmlCodec import com.gzzn.omms.msgexchange.processing.BackfillService +import com.gzzn.omms.msgexchange.processing.OutboundRequestService import com.gzzn.omms.msgexchange.processing.MessageLifecycleGate import org.junit.jupiter.api.Assertions.assertEquals import org.junit.jupiter.api.Assertions.assertNull @@ -36,10 +40,17 @@ class JobRunnerTest { OperationDayProps(), StubPipelineTx(), StubPipelineLock(), ) + private fun reqTrackJob() = ReqTrackMaintenanceJob( + OutboundRequestService( + StubReqTrack(clock), StubCoutmsgOutbox(), JacksonXmlCodec(), props, OperationDayProps(), clock, + ), + ) + private fun runner(proc: ProcStateRepository, activity: JobActivity) = JobRunner( BackfillService(proc, StubInbox(), MailboxProps(), props, clock, MessageLifecycleGate()), historySweep(), EventCleanupJob(StubMsgEvents(), props), ProcStateCleanupJob(proc, MessageLifecycleGate(), props), + reqTrackJob(), clock, activity, props, OperationDayProps(), ) diff --git a/src/test/kotlin/com/gzzn/omms/msgexchange/processing/FlightCommitTest.kt b/src/test/kotlin/com/gzzn/omms/msgexchange/processing/FlightCommitTest.kt index b33898c..c0c2734 100644 --- a/src/test/kotlin/com/gzzn/omms/msgexchange/processing/FlightCommitTest.kt +++ b/src/test/kotlin/com/gzzn/omms/msgexchange/processing/FlightCommitTest.kt @@ -4,6 +4,7 @@ import com.fasterxml.jackson.databind.ObjectMapper import com.gzzn.omms.msgexchange.MutableClock import com.gzzn.omms.msgexchange.codec.DecodeResult import com.gzzn.omms.msgexchange.codec.FlopPayload +import com.gzzn.omms.msgexchange.codec.JacksonXmlCodec import com.gzzn.omms.msgexchange.codec.XmlCodec import com.gzzn.omms.msgexchange.config.OperationDayProps import com.gzzn.omms.msgexchange.config.PipelineProps @@ -205,6 +206,14 @@ class FlightCommitTest { flopProcessor = FlopProcessor(commit, f.flights, f.events, mapper, clock), fdelProcessor = FdelProcessor(commit, f.flights, f.events, mapper, clock), adftProcessor = AdftProcessor(commit, f.flights, f.events, opDay, mapper, clock), + outbound = OutboundRequestService( + com.gzzn.omms.msgexchange.infra.stub.StubReqTrack(clock), + com.gzzn.omms.msgexchange.infra.stub.StubCoutmsgOutbox(), + JacksonXmlCodec(), + props, + opDay, + clock, + ), procFailure = ProcFailure(f.procState, FailureScheduler(props, clock)), props = props, clock = clock, @@ -228,7 +237,8 @@ class FlightCommitTest { private fun codecReturning(decoded: DecodedMessage) = object : XmlCodec { override fun decode(rawXml: String): DecodeResult = DecodeResult.Ok(decoded) - override fun encodeRqrd(kind: String, rangeJson: String): String = "" + override fun encodeOutboundRqfd(seqn: Long, dttm: Long): String = "" + override fun encodeOutboundRqrd(styp: String, seqn: Long, dttm: Long, rtyp: String?): String = "" } private class TracingTx(private val trace: MutableList) : PipelineTransactionManager { diff --git a/src/test/kotlin/com/gzzn/omms/msgexchange/processing/IgnoreBranchTest.kt b/src/test/kotlin/com/gzzn/omms/msgexchange/processing/IgnoreBranchTest.kt index 3f5c5cf..0db742d 100644 --- a/src/test/kotlin/com/gzzn/omms/msgexchange/processing/IgnoreBranchTest.kt +++ b/src/test/kotlin/com/gzzn/omms/msgexchange/processing/IgnoreBranchTest.kt @@ -16,7 +16,12 @@ import com.gzzn.omms.msgexchange.infra.persistence.PipelineTransactionManager import com.gzzn.omms.msgexchange.infra.retry.ProcFailure import com.gzzn.omms.msgexchange.infra.retry.FailureScheduler import com.gzzn.omms.msgexchange.infra.stub.StubFlightState +import com.gzzn.omms.msgexchange.codec.ErorPayload +import com.gzzn.omms.msgexchange.codec.JacksonXmlCodec +import com.gzzn.omms.msgexchange.infra.stub.StubCoutmsgOutbox import com.gzzn.omms.msgexchange.infra.stub.StubInbox +import com.gzzn.omms.msgexchange.infra.stub.StubReqTrack +import com.gzzn.omms.msgexchange.processing.OutboundRequestService import com.gzzn.omms.msgexchange.infra.stub.StubMsgEvents import com.gzzn.omms.msgexchange.infra.stub.StubProcState import com.gzzn.omms.msgexchange.infra.stub.StubSnapshotLog @@ -47,7 +52,8 @@ class IgnoreBranchTest { private fun codecReturning(msg: DecodedMessage) = object : XmlCodec { override fun decode(rawXml: String): DecodeResult = DecodeResult.Ok(msg) - override fun encodeRqrd(kind: String, rangeJson: String): String = "" + override fun encodeOutboundRqfd(seqn: Long, dttm: Long): String = "" + override fun encodeOutboundRqrd(styp: String, seqn: Long, dttm: Long, rtyp: String?): String = "" } private fun processor( @@ -72,6 +78,9 @@ class IgnoreBranchTest { flopProcessor = FlopProcessor(commit, flights, events, ObjectMapper(), clock), fdelProcessor = FdelProcessor(commit, flights, events, ObjectMapper(), clock), adftProcessor = AdftProcessor(commit, flights, events, opDay, ObjectMapper(), clock), + outbound = OutboundRequestService( + StubReqTrack(clock), StubCoutmsgOutbox(), JacksonXmlCodec(), props, opDay, clock, + ), procFailure = procFailure, props = props, clock = clock, @@ -103,15 +112,17 @@ class IgnoreBranchTest { proc.insertIfAbsent(msgId, null) val inbox = StubInbox() inbox.raws[msgId] = "" - val msg = decoded("EROR", "GEN").copy(kind = MsgKind.Eror) + val msg = decoded("EROR", "NONE").copy( + kind = MsgKind.Eror, + body = ErorPayload(99L, "RQFD", "NONE", "test"), + ) val p = processor(proc = proc, inbox = inbox, codec = codecReturning(msg)) p.processOne(head()) val row = proc.find(msgId)!! - assertEquals(ProcStatus.FAILED, row.state) - assertEquals(ErrorClass.UNSUPPORTED, row.errorClass) - assertEquals("eror-pending", row.lastError) + assertEquals(ProcStatus.SKIPPED, row.state) + assertEquals("eror:no-matching-req", row.lastError) } @Test diff --git a/src/test/kotlin/com/gzzn/omms/msgexchange/processing/OutboundRequestTest.kt b/src/test/kotlin/com/gzzn/omms/msgexchange/processing/OutboundRequestTest.kt new file mode 100644 index 0000000..856a5b8 --- /dev/null +++ b/src/test/kotlin/com/gzzn/omms/msgexchange/processing/OutboundRequestTest.kt @@ -0,0 +1,111 @@ +package com.gzzn.omms.msgexchange.processing + +import com.gzzn.omms.msgexchange.MutableClock +import com.gzzn.omms.msgexchange.codec.JacksonXmlCodec +import com.gzzn.omms.msgexchange.config.OperationDayProps +import com.gzzn.omms.msgexchange.config.PipelineProps +import com.gzzn.omms.msgexchange.domain.OutboundRequestKeys +import com.gzzn.omms.msgexchange.infra.persistence.CoutmsgOutboxRepository +import com.gzzn.omms.msgexchange.infra.persistence.ReqTrackRepository +import com.gzzn.omms.msgexchange.infra.stub.StubCoutmsgOutbox +import com.gzzn.omms.msgexchange.infra.stub.StubReqTrack +import org.junit.jupiter.api.Assertions.assertEquals +import org.junit.jupiter.api.Assertions.assertFalse +import org.junit.jupiter.api.Assertions.assertTrue +import org.junit.jupiter.api.Test +import java.time.Duration +import java.time.LocalDate + +class OutboundRequestTest { + + private val clock = MutableClock(MutableClock.BASE) + private val opDay = OperationDayProps().apply { zone = "Asia/Shanghai"; cutoffHour = 0 } + private val props = PipelineProps().apply { outbound.responseTimeout = Duration.ofMinutes(5) } + + private fun service(req: StubReqTrack, outbox: StubCoutmsgOutbox) = + OutboundRequestService(req, outbox, JacksonXmlCodec(), props, opDay, clock) + + @Test + fun `schd sync registers PENDING then confirms COUTMSGS as SENT`() { + val req = StubReqTrack(clock) + val outbox = StubCoutmsgOutbox() + val svc = service(req, outbox) + + val outcome = svc.registerRqfdSync() + assertTrue(outcome is OutboundRequestService.RegisterOutcome.Registered) + val open = req.findLatest( + OutboundRequestKeys.RQFD_REQ_TYPE, + LocalDate.of(2026, 9, 6), + OutboundRequestKeys.SENDER, + listOf(ReqTrackRepository.ReqState.SENT), + )!! + assertEquals(1L, open.coutmsgsId) + assertTrue(outbox.messages.containsValue(outbox.messages.values.first())) + assertTrue(outbox.messages.values.first().contains("OMMS")) + assertTrue(outbox.messages.values.first().contains("RQFD")) + } + + @Test + fun `open slot rejects duplicate schd sync registration`() { + val req = StubReqTrack(clock) + val svc = service(req, StubCoutmsgOutbox()) + svc.registerRqfdSync() + assertEquals(OutboundRequestService.RegisterOutcome.OpenExists, svc.registerRqfdSync()) + } + + @Test + fun `restart retries PENDING without coutmsgs after job dispatch`() { + val req = StubReqTrack(clock) + val outbox = StubCoutmsgOutbox() + val day = LocalDate.of(2026, 9, 6) + val reqId = req.insert(OutboundRequestKeys.RQFD_REQ_TYPE, day, OutboundRequestKeys.SENDER) + assertEquals(ReqTrackRepository.ReqState.PENDING, req.rows[reqId]!!.state) + + val svc = service(req, outbox) + assertEquals(1, svc.dispatchPending()) + + assertEquals(ReqTrackRepository.ReqState.SENT, req.rows[reqId]!!.state) + } + + @Test + fun `ambiguous write marks uncertain and does not blind resend`() { + val req = StubReqTrack(clock) + val outbox = StubCoutmsgOutbox().apply { nextInsertOutcome = CoutmsgOutboxRepository.InsertOutcome.AMBIGUOUS } + val day = LocalDate.of(2026, 9, 6) + val reqId = req.insert(OutboundRequestKeys.RQFD_REQ_TYPE, day, OutboundRequestKeys.SENDER) + val svc = service(req, outbox) + + assertFalse(svc.dispatchPending() > 0) + assertTrue(req.rows[reqId]!!.writeUncertain) + assertEquals(ReqTrackRepository.ReqState.PENDING, req.rows[reqId]!!.state) + outbox.nextInsertOutcome = CoutmsgOutboxRepository.InsertOutcome.CONFIRMED + assertEquals(0, svc.dispatchPending()) + } + + @Test + fun `timeout expires SENT request`() { + val req = StubReqTrack(clock) + val day = LocalDate.of(2026, 9, 6) + val reqId = req.insert(OutboundRequestKeys.RQFD_REQ_TYPE, day, OutboundRequestKeys.SENDER) + req.linkCoutmsgs(reqId, 10L, 99L) + req.markSent(reqId, clock.instant()) + clock.advance(Duration.ofMinutes(10).toMillis()) + + val svc = service(req, StubCoutmsgOutbox()) + assertEquals(1, svc.expireTimedOut()) + assertEquals(ReqTrackRepository.ReqState.EXPIRED, req.rows[reqId]!!.state) + } + + @Test + fun `EROR fails matching open outbound by seqn`() { + val req = StubReqTrack(clock) + val day = LocalDate.of(2026, 9, 6) + val reqId = req.insert(OutboundRequestKeys.RQFD_REQ_TYPE, day, OutboundRequestKeys.SENDER) + req.linkCoutmsgs(reqId, 1L, 42L) + req.markSent(reqId, clock.instant()) + + val svc = service(req, StubCoutmsgOutbox()) + assertTrue(svc.failFromEror(42L, "RQFD", "NONE")) + assertEquals(ReqTrackRepository.ReqState.FAILED, req.rows[reqId]!!.state) + } +} diff --git a/src/test/kotlin/com/gzzn/omms/msgexchange/processing/RespGuardTest.kt b/src/test/kotlin/com/gzzn/omms/msgexchange/processing/RespGuardTest.kt new file mode 100644 index 0000000..a5c6f0c --- /dev/null +++ b/src/test/kotlin/com/gzzn/omms/msgexchange/processing/RespGuardTest.kt @@ -0,0 +1,127 @@ +package com.gzzn.omms.msgexchange.processing + +import com.fasterxml.jackson.databind.ObjectMapper +import com.gzzn.omms.msgexchange.codec.JacksonXmlCodec +import com.gzzn.omms.msgexchange.codec.ScheduleBody +import com.gzzn.omms.msgexchange.config.OperationDayProps +import com.gzzn.omms.msgexchange.config.PipelineProps +import com.gzzn.omms.msgexchange.domain.DecodedMessage +import com.gzzn.omms.msgexchange.domain.MetaFields +import com.gzzn.omms.msgexchange.domain.MsgKind +import com.gzzn.omms.msgexchange.domain.OutboundRequestKeys +import com.gzzn.omms.msgexchange.domain.ProcState +import com.gzzn.omms.msgexchange.domain.ProcStatus +import com.gzzn.omms.msgexchange.infra.metrics.PipelineCounters +import com.gzzn.omms.msgexchange.infra.persistence.ReqTrackRepository +import com.gzzn.omms.msgexchange.infra.retry.FailureScheduler +import com.gzzn.omms.msgexchange.infra.retry.ProcFailure +import com.gzzn.omms.msgexchange.infra.stub.StubCoutmsgOutbox +import com.gzzn.omms.msgexchange.infra.stub.StubFlightProjectionPort +import com.gzzn.omms.msgexchange.infra.stub.StubFlightState +import com.gzzn.omms.msgexchange.infra.stub.StubInbox +import com.gzzn.omms.msgexchange.infra.stub.StubMsgEvents +import com.gzzn.omms.msgexchange.infra.stub.StubProcState +import com.gzzn.omms.msgexchange.infra.stub.StubReqTrack +import com.gzzn.omms.msgexchange.infra.stub.StubSnapshotLog +import com.gzzn.omms.msgexchange.MutableClock +import org.junit.jupiter.api.Assertions.assertEquals +import org.junit.jupiter.api.Test +import java.time.Instant +import java.time.LocalDate + +class RespGuardTest { + + private val clock = MutableClock(MutableClock.BASE) + private val msgId = 900L + private val day = LocalDate.of(2026, 9, 6) + + private fun schdMsg(styp: MsgKind.SchdSubtype, body: ScheduleBody) = DecodedMessage( + meta = MetaFields("AODB", "SCHD", styp.name, 1L, 1L), + kind = MsgKind.Schd(styp), + rawXml = "", + body = body, + ) + + private fun emptyBody() = ScheduleBody(0, emptyList()) + + private fun processor( + proc: StubProcState, + inbox: StubInbox, + req: StubReqTrack, + decoded: DecodedMessage, + ): MessageProcessor { + val props = PipelineProps() + val opDay = OperationDayProps().apply { zone = "Asia/Shanghai"; cutoffHour = 0 } + val flights = StubFlightState() + val events = StubMsgEvents() + val commit = testCommit(procState = proc, msgEvents = events, projection = StubFlightProjectionPort(), clock = clock) + return MessageProcessor( + inbox = inbox, + procState = proc, + codec = object : com.gzzn.omms.msgexchange.codec.XmlCodec { + override fun decode(rawXml: String) = + com.gzzn.omms.msgexchange.codec.DecodeResult.Ok(decoded) + override fun encodeOutboundRqfd(seqn: Long, dttm: Long) = "" + override fun encodeOutboundRqrd(styp: String, seqn: Long, dttm: Long, rtyp: String?) = "" + }, + scheduleProcessor = ScheduleProcessor( + commit, proc, flights, events, StubSnapshotLog(), opDay, props, ObjectMapper(), clock, + ), + flopProcessor = FlopProcessor(commit, flights, events, ObjectMapper(), clock), + fdelProcessor = FdelProcessor(commit, flights, events, ObjectMapper(), clock), + adftProcessor = AdftProcessor(commit, flights, events, opDay, ObjectMapper(), clock), + outbound = OutboundRequestService(req, StubCoutmsgOutbox(), JacksonXmlCodec(), props, opDay, clock), + procFailure = ProcFailure(proc, FailureScheduler(props, clock)), + props = props, + clock = clock, + operationDayProps = opDay, + counters = PipelineCounters(), + ) + } + + @Test + fun `SCHD-RESP without open RQFD is SKIPPED`() { + val proc = StubProcState() + proc.insertIfAbsent(msgId, null) + val inbox = StubInbox() + inbox.raws[msgId] = "" + val p = processor(proc, inbox, StubReqTrack(clock), schdMsg(MsgKind.SchdSubtype.RESP, emptyBody())) + + p.processOne(ProcState(msgId, ProcStatus.PENDING, updatedAt = Instant.EPOCH)) + + assertEquals(ProcStatus.SKIPPED, proc.find(msgId)!!.state) + assertEquals("resp-guard:no-open-req", proc.find(msgId)!!.lastError) + } + + @Test + fun `SCHD-DNLD does not require open request`() { + val proc = StubProcState() + proc.insertIfAbsent(msgId, null) + val inbox = StubInbox() + inbox.raws[msgId] = "" + val p = processor(proc, inbox, StubReqTrack(clock), schdMsg(MsgKind.SchdSubtype.DNLD, emptyBody())) + + p.processOne(ProcState(msgId, ProcStatus.PENDING, updatedAt = Instant.EPOCH)) + + assertEquals(ProcStatus.SUCCEEDED, proc.find(msgId)!!.state) + } + + @Test + fun `duplicate SCHD-RESP after completion is SKIPPED`() { + val req = StubReqTrack(clock) + val reqId = req.insert(OutboundRequestKeys.RQFD_REQ_TYPE, day, OutboundRequestKeys.SENDER) + req.linkCoutmsgs(reqId, 1L, 1L) + req.markSent(reqId, clock.instant()) + req.completeLatest(OutboundRequestKeys.RQFD_REQ_TYPE, day, OutboundRequestKeys.SENDER) + assertEquals(ReqTrackRepository.ReqState.DONE, req.rows[reqId]!!.state) + + val proc = StubProcState() + proc.insertIfAbsent(msgId, null) + val inbox = StubInbox() + inbox.raws[msgId] = "" + val p = processor(proc, inbox, req, schdMsg(MsgKind.SchdSubtype.RESP, emptyBody())) + p.processOne(ProcState(msgId, ProcStatus.PENDING, updatedAt = Instant.EPOCH)) + + assertEquals(ProcStatus.SKIPPED, proc.find(msgId)!!.state) + } +}