diff --git a/common-core/src/main/java/com/baiyue/common/core/outbox/OutboxOperations.java b/common-core/src/main/java/com/baiyue/common/core/outbox/OutboxOperations.java index cdeaf1f48..f6f0664a5 100644 --- a/common-core/src/main/java/com/baiyue/common/core/outbox/OutboxOperations.java +++ b/common-core/src/main/java/com/baiyue/common/core/outbox/OutboxOperations.java @@ -7,6 +7,7 @@ import org.springframework.jdbc.core.JdbcTemplate; import org.springframework.transaction.PlatformTransactionManager; import org.springframework.transaction.support.TransactionTemplate; import java.time.LocalDateTime; +import java.time.ZoneOffset; import java.util.List; import java.util.Map; import java.util.Set; @@ -76,7 +77,8 @@ public final class OutboxOperations { return result(canonicalId, identity, request, true); } if (!"DEAD".equals(row.get("entry_status"))) throw conflict("仅允许重试 DEAD 事件"); - LocalDateTime now = LocalDateTime.now(); + // Relay due-time comparisons use UTC, independently of the JVM default zone. + LocalDateTime now = LocalDateTime.now(ZoneOffset.UTC); int updated = jdbc.update("UPDATE " + layout.table() + " SET " + layout.statusColumn() + "='PENDING',retry_count=0,next_retry_at=?,locked_by=NULL,locked_at=NULL,last_error=NULL WHERE " + layout.idColumn() + "=? AND " + layout.statusColumn() + "='DEAD'", now, canonicalId); diff --git a/common-core/src/test/java/com/baiyue/common/core/OutboxOperationsTest.java b/common-core/src/test/java/com/baiyue/common/core/OutboxOperationsTest.java index 43096352d..2570a4853 100644 --- a/common-core/src/test/java/com/baiyue/common/core/OutboxOperationsTest.java +++ b/common-core/src/test/java/com/baiyue/common/core/OutboxOperationsTest.java @@ -5,15 +5,20 @@ import com.baiyue.common.core.outbox.OutboxOperations; import org.h2.jdbcx.JdbcDataSource; import org.junit.jupiter.api.BeforeEach; import org.junit.jupiter.api.Test; +import org.junit.jupiter.api.parallel.Isolated; import org.springframework.jdbc.core.JdbcTemplate; import org.springframework.jdbc.datasource.DataSourceTransactionManager; +import java.time.LocalDateTime; +import java.time.ZoneOffset; import java.util.Set; +import java.util.TimeZone; import java.util.UUID; import java.util.concurrent.CountDownLatch; import java.util.concurrent.Executors; import java.util.concurrent.TimeUnit; import static org.assertj.core.api.Assertions.*; +@Isolated("UTC retry regression temporarily changes the JVM default time zone") class OutboxOperationsTest { private static final String HASH = "sha256:" + "a".repeat(64); private JdbcTemplate jdbc; @@ -54,6 +59,35 @@ class OutboxOperationsTest { assertThat(jdbc.queryForObject("SELECT COUNT(*) FROM event_delivery_retry_audit", Integer.class)).isZero(); } + @Test void retryIsImmediatelyDueInUtcUnderShanghaiDefaultZoneAndReplayKeepsTimes() { + TimeZone original = TimeZone.getDefault(); + try { + TimeZone.setDefault(TimeZone.getTimeZone("Asia/Shanghai")); + var request = request(); + LocalDateTime before = LocalDateTime.now(ZoneOffset.UTC).minusSeconds(1); + assertThat(operations.retry("1", request).duplicate()).isFalse(); + LocalDateTime nextRetry = jdbc.queryForObject("SELECT next_retry_at FROM events WHERE id='1'", + (row, index) -> row.getObject(1, LocalDateTime.class)); + LocalDateTime requested = jdbc.queryForObject("SELECT requested_at FROM event_delivery_retry_audit", + (row, index) -> row.getObject(1, LocalDateTime.class)); + LocalDateTime scan = LocalDateTime.now(ZoneOffset.UTC); + // Allow database timestamp precision rounding, but never a default-zone offset. + assertThat(nextRetry).isBetween(before, scan.plusSeconds(1)); + assertThat(requested).isEqualTo(nextRetry); + assertThat(jdbc.queryForObject("SELECT COUNT(*) FROM events WHERE id='1' AND status='PENDING' AND next_retry_at<=?", + Integer.class, scan)).isEqualTo(1); + assertThat(operations.retry("1", request).duplicate()).isTrue(); + assertThat(jdbc.queryForObject("SELECT next_retry_at FROM events WHERE id='1'", + (row, index) -> row.getObject(1, LocalDateTime.class))).isEqualTo(nextRetry); + assertThat(jdbc.queryForObject("SELECT requested_at FROM event_delivery_retry_audit", + (row, index) -> row.getObject(1, LocalDateTime.class))).isEqualTo(requested); + assertThat(jdbc.queryForObject("SELECT COUNT(*) FROM event_delivery_retry_audit", Integer.class)).isEqualTo(1); + assertThat(jdbc.queryForObject("SELECT payload_json FROM events WHERE id='1'", String.class)).isEqualTo("event-1"); + } finally { + TimeZone.setDefault(original); + } + } + @Test void staleHashAndNonDeadAndReusedRequestAreRejected() { var original = request(); var stale = new OutboxOperations.RetryRequest(original.requestId(), original.operator(), original.reason(), "sha256:" + "b".repeat(64)); diff --git a/lastleg-service/src/test/java/com/baiyue/lastleg/financefee/OverseasFeeCancellationPersistenceTest.java b/lastleg-service/src/test/java/com/baiyue/lastleg/financefee/OverseasFeeCancellationPersistenceTest.java index 75e059060..0141b02e0 100644 --- a/lastleg-service/src/test/java/com/baiyue/lastleg/financefee/OverseasFeeCancellationPersistenceTest.java +++ b/lastleg-service/src/test/java/com/baiyue/lastleg/financefee/OverseasFeeCancellationPersistenceTest.java @@ -4,6 +4,9 @@ import com.baomidou.mybatisplus.core.MybatisConfiguration; import com.baomidou.mybatisplus.extension.spring.MybatisSqlSessionFactoryBean; import com.baiyue.common.core.outbox.EventDeliveryJournal; import com.baiyue.common.core.outbox.EventDeliveryJournal.Outcome; +import com.baiyue.common.core.outbox.OutboxOperations; +import com.baiyue.common.event.EventEnvelope; +import com.baiyue.common.event.EventTypes; import com.baiyue.lastleg.entity.LastlegEventOutbox; import com.baiyue.lastleg.financefee.entity.OverseasFeeSource; import com.baiyue.lastleg.financefee.mapper.OverseasFeeSourceMapper; @@ -14,16 +17,21 @@ import com.baiyue.lastleg.mapper.LastlegEventOutboxMapper; import com.fasterxml.jackson.databind.ObjectMapper; import org.junit.jupiter.api.BeforeEach; import org.junit.jupiter.api.Test; +import org.junit.jupiter.api.parallel.Isolated; import org.mybatis.spring.SqlSessionTemplate; import org.springframework.jdbc.core.JdbcTemplate; import org.springframework.jdbc.datasource.DataSourceTransactionManager; import org.springframework.jdbc.datasource.DriverManagerDataSource; import org.springframework.test.util.ReflectionTestUtils; import java.time.LocalDateTime; +import java.time.ZoneOffset; import java.util.List; +import java.util.Set; +import java.util.TimeZone; import java.util.UUID; import static org.assertj.core.api.Assertions.*; +@Isolated("Manual retry regression temporarily changes the JVM default time zone") class OverseasFeeCancellationPersistenceTest { JdbcTemplate jdbc; LastlegEventOutboxMapper events; @@ -31,6 +39,7 @@ class OverseasFeeCancellationPersistenceTest { OverseasFeeCancellation cancellation; OverseasFeeOutboxRepository repository; EventDeliveryJournal journal; + OutboxOperations operations; OverseasFeeSource source; LastlegEventOutbox request; final LocalDateTime now = LocalDateTime.of(2026, 10, 3, 1, 0); @@ -39,6 +48,7 @@ class OverseasFeeCancellationPersistenceTest { jdbc = new JdbcTemplate(dataSource); jdbc.execute("CREATE TABLE ll_event_outbox(id BIGINT PRIMARY KEY,customer_id VARCHAR(64),aggregate_type VARCHAR(64),aggregate_id VARCHAR(64),event_version INT,event_type VARCHAR(96),event_key VARCHAR(200) UNIQUE,payload_json TEXT,status VARCHAR(32),retry_count INT NOT NULL DEFAULT 0,next_retry_at TIMESTAMP,published_at TIMESTAMP,locked_at TIMESTAMP,locked_by VARCHAR(128),last_error VARCHAR(1000),created_at TIMESTAMP)"); jdbc.execute("CREATE TABLE event_delivery_attempt(attempt_id VARCHAR(36) PRIMARY KEY,service_name VARCHAR(32) NOT NULL,event_type VARCHAR(96) NOT NULL,entry_id VARCHAR(128) NOT NULL,event_id VARCHAR(128),target_name VARCHAR(64) NOT NULL,started_at TIMESTAMP(3) NOT NULL,finished_at TIMESTAMP(3),duration_ms BIGINT,outcome VARCHAR(24) NOT NULL,error_code VARCHAR(64),next_retry_at TIMESTAMP(3))"); + jdbc.execute("CREATE TABLE event_delivery_retry_audit(request_id VARCHAR(36) PRIMARY KEY,producer_table VARCHAR(100),entry_id VARCHAR(64),event_id VARCHAR(64),event_type VARCHAR(100),payload_hash VARCHAR(71),operator_name VARCHAR(128),reason VARCHAR(500),previous_failures INT,previous_error VARCHAR(1000),requested_at TIMESTAMP)"); jdbc.execute("CREATE TABLE ll_finance_fee_source(id BIGINT PRIMARY KEY,source_fee_id BIGINT,source_version INT,customer_id VARCHAR(64),source_type VARCHAR(64),source_status VARCHAR(32),updated_at TIMESTAMP,business_type VARCHAR(64),direction VARCHAR(32),fee_item_code VARCHAR(64),inbound_code VARCHAR(64),review_status VARCHAR(32))"); jdbc.update("INSERT INTO ll_finance_fee_source(id,source_fee_id,source_version,customer_id,source_type,source_status) VALUES(10,701,1,'10001','OVERSEAS_DELIVERY_FEE','ACTIVE')"); var config = new MybatisConfiguration(); config.addMapper(LastlegEventOutboxMapper.class); config.addMapper(OverseasFeeSourceMapper.class); @@ -49,6 +59,14 @@ class OverseasFeeCancellationPersistenceTest { cancellation = new OverseasFeeCancellation(events, fees, json); repository = new OverseasFeeOutboxRepository(events, cancellation); journal = new EventDeliveryJournal(jdbc, new DataSourceTransactionManager(dataSource)); + operations = new OutboxOperations(jdbc, new DataSourceTransactionManager(dataSource), + new OutboxOperations.Layout("ll_event_outbox", "id", "status", Set.of(EventTypes.BUSINESS_FEE_SOURCE_CHANGED)), + payload -> { + try { + var envelope = json.readValue(payload, EventEnvelope.class).requireValid(); + return new OutboxOperations.Identity(envelope.eventId(), envelope.eventType(), envelope.payloadHash()); + } catch (Exception error) { throw new IllegalStateException(error); } + }); source = fees.lockCurrent(701); var producer = new OverseasFeeOutboxService(); ReflectionTestUtils.setField(producer, "mapper", events); ReflectionTestUtils.setField(producer, "objectMapper", json); @@ -73,6 +91,41 @@ class OverseasFeeCancellationPersistenceTest { assertThat(ready).hasSize(1); assertThat(ready.get(0).getEventKey()).isEqualTo(OverseasFeeCancellation.eventKey(source, "PUBLISHED")); } + @Test void manualRetryInShanghaiIsImmediatelySelectableAndClaimableByUtcFeeRelay() { + TimeZone original = TimeZone.getDefault(); + try { + TimeZone.setDefault(TimeZone.getTimeZone("Asia/Shanghai")); + publishedOriginal(); + jdbc.update("UPDATE ll_event_outbox SET status='DEAD',retry_count=12,last_error='ACK lost' WHERE id=?", request.getId()); + String entryId = String.valueOf(request.getId()); + var identity = operations.inspect(entryId); + var retry = new OutboxOperations.RetryRequest(UUID.randomUUID().toString(), "test-operator", "ACK restored", identity.payloadHash()); + assertThat(operations.retry(entryId, retry).duplicate()).isFalse(); + LocalDateTime nextRetry = events.selectById(request.getId()).getNextRetryAt(); + LocalDateTime requested = jdbc.queryForObject("SELECT requested_at FROM event_delivery_retry_audit", + (row, index) -> row.getObject(1, LocalDateTime.class)); + assertThat(operations.retry(entryId, retry).duplicate()).isTrue(); + assertThat(events.selectById(request.getId()).getNextRetryAt()).isEqualTo(nextRetry); + assertThat(jdbc.queryForObject("SELECT requested_at FROM event_delivery_retry_audit", + (row, index) -> row.getObject(1, LocalDateTime.class))).isEqualTo(requested); + assertThat(jdbc.queryForObject("SELECT COUNT(*) FROM event_delivery_retry_audit", Integer.class)).isEqualTo(1); + LocalDateTime scan = LocalDateTime.now(ZoneOffset.UTC); + var ready = repository.pending(50, scan, scan.minusSeconds(60)); + assertThat(ready).extracting(LastlegEventOutbox::getId).containsExactly(request.getId()); + assertThat(ready.get(0).getPayloadJson()).isEqualTo(request.getPayloadJson()); + assertThat(repository.prerequisitePublished(ready.get(0))).isTrue(); + assertThat(journal.begin("utc-worker", "lastleg-service", request.getEventType(), entryId, identity.eventId(), "finance", scan, + () -> repository.claim(request.getId(), "utc-worker", scan))).isTrue(); + assertThat(repository.claimed(request.getId(), "utc-worker").getPayloadJson()).isEqualTo(request.getPayloadJson()); + assertThat(journal.complete("utc-worker", Outcome.DUPLICATE, scan, null, null, + () -> repository.published(request.getId(), "utc-worker", scan))).isTrue(); + assertThat(events.selectById(request.getId()).getStatus()).isEqualTo("PUBLISHED"); + assertThat(events.selectById(request.getId()).getPayloadJson()).isEqualTo(request.getPayloadJson()); + assertThat(status()).isEqualTo("VOIDED"); + } finally { + TimeZone.setDefault(original); + } + } @Test void journalFailureRollsBackBothLocalVoidAndOutboxCompletionThenSameAckCanCommit() { publishedOriginal(); claim(); jdbc.execute("ALTER TABLE event_delivery_attempt ADD CONSTRAINT reject_applied CHECK(outcome <> 'APPLIED')");