java-project/baiyue-firstleg
4
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

fix: schedule manual outbox retries in UTC

Browse Source
This commit is contained in:
ouyangxq
2026-10-03 09:27:46 +08:00
parent 588227f181
commit d93185e750
3 changed files with 90 additions and 1 deletions
Download Patch File Download Diff File Expand all files Collapse all files
common-core/src/main/java/com/baiyue/common/core/outbox/OutboxOperations.java
+3 -1
View File
@@ -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);
common-core/src/test/java/com/baiyue/common/core/OutboxOperationsTest.java
+34
View File
@@ -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.<LocalDateTime>queryForObject("SELECT next_retry_at FROM events WHERE id='1'",
(row, index) -> row.getObject(1, LocalDateTime.class))).isEqualTo(nextRetry);
assertThat(jdbc.<LocalDateTime>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));
lastleg-service/src/test/java/com/baiyue/lastleg/financefee/OverseasFeeCancellationPersistenceTest.java
+53
View File
@@ -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.<LocalDateTime>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')");