优化
This commit is contained in:
@@ -15,6 +15,7 @@
|
||||
| 通用审计字段 | `src/main/java/com/baiyue/common/core/model/BaseEntity.java` |
|
||||
| MyBatis-Plus 与审计人 | `src/main/java/com/baiyue/common/core/mybatis/` |
|
||||
| 有限重试与退避 | `src/main/java/com/baiyue/common/core/retry/BoundedRetryPolicy.java`；供多个服务的 Outbox 复用，不拥有持久化或投递逻辑 |
|
||||
| 人工重试事务与凭据 | `src/main/java/com/baiyue/common/core/outbox/OutboxOperations.java`、`EventOperationsTokenGuard.java`；三个生产服务显式选择本库表及事件白名单，锁定 DEAD 原记录，状态恢复与审计同事务；不访问其他服务数据库或直接投递事件 |
|
||||
| 响应契约测试 | `src/test/java/com/baiyue/common/core/ApiResponseTest.java` |
|
||||
|
||||
## 修改与验证
|
||||
|
||||
@@ -24,5 +24,7 @@
|
||||
<artifactId>spring-boot-starter-test</artifactId>
|
||||
<scope>test</scope>
|
||||
</dependency>
|
||||
<dependency><groupId>org.springframework</groupId><artifactId>spring-jdbc</artifactId></dependency>
|
||||
<dependency><groupId>com.h2database</groupId><artifactId>h2</artifactId><scope>test</scope></dependency>
|
||||
</dependencies>
|
||||
</project>
|
||||
|
||||
@@ -0,0 +1,59 @@
|
||||
package com.baiyue.common.core.outbox;
|
||||
|
||||
import org.springframework.jdbc.core.JdbcTemplate;
|
||||
import org.springframework.jdbc.datasource.DataSourceTransactionManager;
|
||||
import org.springframework.transaction.PlatformTransactionManager;
|
||||
import org.springframework.transaction.TransactionDefinition;
|
||||
import org.springframework.transaction.support.TransactionTemplate;
|
||||
import java.time.Duration;
|
||||
import java.time.LocalDateTime;
|
||||
import java.util.Objects;
|
||||
import java.util.Set;
|
||||
import java.util.function.BooleanSupplier;
|
||||
import java.util.function.IntSupplier;
|
||||
|
||||
/** Same-database audit. HTTP must only run after begin returns; never hold a DB transaction over HTTP. */
|
||||
public final class EventDeliveryJournal {
|
||||
public enum Outcome { APPLIED, DUPLICATE, STALE, RETRY, DEAD, LEASE_EXPIRED }
|
||||
private final JdbcTemplate jdbc;
|
||||
private final TransactionTemplate transaction;
|
||||
|
||||
public EventDeliveryJournal(JdbcTemplate jdbc, PlatformTransactionManager manager) {
|
||||
this.jdbc = Objects.requireNonNull(jdbc);
|
||||
if (!(manager instanceof DataSourceTransactionManager local)
|
||||
|| local.getDataSource() != jdbc.getDataSource()) {
|
||||
throw new IllegalArgumentException("Delivery audit must share the local Outbox transaction manager");
|
||||
}
|
||||
transaction = new TransactionTemplate(manager);
|
||||
transaction.setPropagationBehavior(TransactionDefinition.PROPAGATION_REQUIRES_NEW);
|
||||
}
|
||||
|
||||
public boolean begin(String attempt, String service, String eventType, String entryId,
|
||||
String eventId, String target, LocalDateTime now, BooleanSupplier claim) {
|
||||
return Boolean.TRUE.equals(transaction.execute(status -> {
|
||||
if (!claim.getAsBoolean()) return false;
|
||||
jdbc.update("INSERT INTO event_delivery_attempt(attempt_id,service_name,event_type,entry_id,event_id,target_name,started_at,outcome) VALUES(?,?,?,?,?,?,?,'STARTED')",
|
||||
attempt, service, eventType, entryId, eventId, target, now);
|
||||
return true;
|
||||
}));
|
||||
}
|
||||
|
||||
public boolean complete(String attempt, Outcome outcome, LocalDateTime now,
|
||||
String errorCode, LocalDateTime nextRetry, IntSupplier updateOwnedOutbox) {
|
||||
if (errorCode != null && !Set.of("RETRY_EXHAUSTED", "LEASE_EXPIRED", "INVALID_EVENT", "PERMANENT_FAILURE", "DELIVERY_FAILURE").contains(errorCode)
|
||||
&& !errorCode.matches("HTTP_[1-5][0-9]{2}")) throw new IllegalArgumentException("Unbounded audit error code");
|
||||
return Boolean.TRUE.equals(transaction.execute(status -> {
|
||||
int changed = updateOwnedOutbox.getAsInt();
|
||||
if (changed == 0) return false; // A stale worker has no right to change either record.
|
||||
if (changed != 1) throw new IllegalStateException("Unexpected Outbox update count");
|
||||
LocalDateTime started = jdbc.queryForObject(
|
||||
"SELECT started_at FROM event_delivery_attempt WHERE attempt_id=? AND outcome='STARTED' FOR UPDATE",
|
||||
LocalDateTime.class, attempt);
|
||||
long duration = Math.max(0, Duration.between(Objects.requireNonNull(started), now).toMillis());
|
||||
int audited = jdbc.update("UPDATE event_delivery_attempt SET finished_at=?,duration_ms=?,outcome=?,error_code=?,next_retry_at=? WHERE attempt_id=? AND outcome='STARTED'",
|
||||
now, duration, outcome.name(), errorCode, nextRetry, attempt);
|
||||
if (audited != 1) throw new IllegalStateException("Delivery completion audit is missing");
|
||||
return true;
|
||||
}));
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,4 @@
|
||||
package com.baiyue.common.core.outbox;
|
||||
|
||||
/** Marks services that require durable attempt audit and cached delivery metrics. */
|
||||
public interface EventDeliveryProducer { }
|
||||
+20
@@ -0,0 +1,20 @@
|
||||
package com.baiyue.common.core.outbox;
|
||||
|
||||
import com.baiyue.common.core.api.ResultCode;
|
||||
import com.baiyue.common.core.exception.BizException;
|
||||
import java.nio.charset.StandardCharsets;
|
||||
import java.security.MessageDigest;
|
||||
|
||||
/** Dedicated operations credential; a missing or weak configuration disables this capability. */
|
||||
public final class EventOperationsTokenGuard {
|
||||
private final byte[] expected;
|
||||
public EventOperationsTokenGuard(String token) {
|
||||
expected = token == null ? new byte[0] : token.getBytes(StandardCharsets.UTF_8);
|
||||
}
|
||||
public void require(String supplied) {
|
||||
if (expected.length < 32 || supplied == null || supplied.length() > 4096
|
||||
|| !MessageDigest.isEqual(expected, supplied.getBytes(StandardCharsets.UTF_8))) {
|
||||
throw new BizException(ResultCode.FORBIDDEN, "事件运维凭据无效或入口未启用");
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,129 @@
|
||||
package com.baiyue.common.core.outbox;
|
||||
|
||||
import com.baiyue.common.core.api.ResultCode;
|
||||
import com.baiyue.common.core.exception.BizException;
|
||||
import org.springframework.dao.DuplicateKeyException;
|
||||
import org.springframework.jdbc.core.JdbcTemplate;
|
||||
import org.springframework.transaction.PlatformTransactionManager;
|
||||
import org.springframework.transaction.support.TransactionTemplate;
|
||||
import java.time.LocalDateTime;
|
||||
import java.util.List;
|
||||
import java.util.Map;
|
||||
import java.util.Set;
|
||||
import java.util.UUID;
|
||||
import java.util.function.Function;
|
||||
|
||||
/** Transactional operator retry of an immutable, service-owned Outbox entry. No cross-schema access. */
|
||||
public final class OutboxOperations {
|
||||
public record Layout(String table, String idColumn, String statusColumn, Set<String> eventTypes) {
|
||||
public Layout {
|
||||
for (String name : List.of(table, idColumn, statusColumn)) {
|
||||
if (!name.matches("[a-z][a-z0-9_]*")) throw new IllegalArgumentException("Invalid Outbox identifier");
|
||||
}
|
||||
eventTypes = Set.copyOf(eventTypes);
|
||||
if (eventTypes.isEmpty()) throw new IllegalArgumentException("Event allowlist is required");
|
||||
}
|
||||
}
|
||||
public record Identity(String eventId, String eventType, String payloadHash) { }
|
||||
public record RetryRequest(String requestId, String operator, String reason, String expectedPayloadHash) { }
|
||||
public record RetryResult(String entryId, String eventId, String payloadHash, String requestId, boolean duplicate) { }
|
||||
public record Entry(String entryId, String eventId, String eventType, String payloadHash,
|
||||
String status, int failures, String lastError, Object nextRetryAt) { }
|
||||
private final JdbcTemplate jdbc;
|
||||
private final Layout layout;
|
||||
private final Function<String, Identity> identityReader;
|
||||
private final TransactionTemplate transaction;
|
||||
|
||||
public OutboxOperations(JdbcTemplate jdbc, PlatformTransactionManager transactions, Layout layout,
|
||||
Function<String, Identity> identityReader) {
|
||||
this.jdbc = jdbc;
|
||||
this.layout = layout;
|
||||
this.identityReader = identityReader;
|
||||
this.transaction = new TransactionTemplate(transactions);
|
||||
}
|
||||
|
||||
public Entry inspect(String entryId) {
|
||||
validateId(entryId);
|
||||
Map<String, Object> row = find(entryId, false);
|
||||
Identity identity = identity(row);
|
||||
return new Entry(String.valueOf(row.get("entry_id")), identity.eventId(), identity.eventType(), identity.payloadHash(),
|
||||
String.valueOf(row.get("entry_status")), ((Number) row.get("retry_count")).intValue(),
|
||||
(String) row.get("last_error"), row.get("next_retry_at"));
|
||||
}
|
||||
|
||||
public RetryResult retry(String entryId, RetryRequest request) {
|
||||
validateId(entryId);
|
||||
validateRequest(request);
|
||||
try {
|
||||
return transaction.execute(status -> {
|
||||
Map<String, Object> row = find(entryId, true);
|
||||
String canonicalId = String.valueOf(row.get("entry_id"));
|
||||
Identity identity = identity(row);
|
||||
if (!identity.payloadHash().equalsIgnoreCase(request.expectedPayloadHash())) {
|
||||
throw conflict("事件摘要已变化，拒绝重试");
|
||||
}
|
||||
List<Map<String, Object>> prior = jdbc.queryForList(
|
||||
"SELECT producer_table,entry_id,event_id,payload_hash,operator_name,reason FROM event_delivery_retry_audit WHERE request_id=?",
|
||||
request.requestId());
|
||||
if (!prior.isEmpty()) {
|
||||
Map<String, Object> audit = prior.get(0);
|
||||
if (!layout.table().equals(audit.get("producer_table")) || !canonicalId.equals(audit.get("entry_id"))
|
||||
|| !identity.eventId().equals(audit.get("event_id"))
|
||||
|| !identity.payloadHash().equals(audit.get("payload_hash"))
|
||||
|| !request.operator().equals(audit.get("operator_name")) || !request.reason().equals(audit.get("reason"))) {
|
||||
throw conflict("操作请求编号已用于不同重试参数");
|
||||
}
|
||||
return result(canonicalId, identity, request, true);
|
||||
}
|
||||
if (!"DEAD".equals(row.get("entry_status"))) throw conflict("仅允许重试 DEAD 事件");
|
||||
LocalDateTime now = LocalDateTime.now();
|
||||
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);
|
||||
if (updated != 1) throw conflict("事件状态已变化");
|
||||
jdbc.update("INSERT INTO event_delivery_retry_audit(request_id,producer_table,entry_id,event_id,event_type,payload_hash,operator_name,reason,previous_failures,previous_error,requested_at) VALUES(?,?,?,?,?,?,?,?,?,?,?)",
|
||||
request.requestId(), layout.table(), canonicalId, identity.eventId(), identity.eventType(), identity.payloadHash(),
|
||||
request.operator(), request.reason(), row.get("retry_count"), row.get("last_error"), now);
|
||||
return result(canonicalId, identity, request, false);
|
||||
});
|
||||
} catch (DuplicateKeyException duplicate) {
|
||||
throw conflict("操作请求编号冲突，原事务已回滚");
|
||||
}
|
||||
}
|
||||
|
||||
private Map<String, Object> find(String id, boolean lock) {
|
||||
List<Map<String, Object>> rows = jdbc.queryForList("SELECT " + layout.idColumn() + " AS entry_id,payload_json," + layout.statusColumn()
|
||||
+ " AS entry_status,retry_count,last_error,next_retry_at FROM " + layout.table() + " WHERE "
|
||||
+ layout.idColumn() + "=?" + (lock ? " FOR UPDATE" : ""), id);
|
||||
if (rows.isEmpty()) throw new BizException(ResultCode.NOT_FOUND, "事件不存在");
|
||||
return rows.get(0);
|
||||
}
|
||||
private Identity identity(Map<String, Object> row) {
|
||||
Identity identity = identityReader.apply(String.valueOf(row.get("payload_json")));
|
||||
if (!layout.eventTypes().contains(identity.eventType())) throw new BizException(ResultCode.FORBIDDEN, "事件不属于本运维入口");
|
||||
return identity;
|
||||
}
|
||||
private static RetryResult result(String entryId, Identity identity, RetryRequest request, boolean duplicate) {
|
||||
return new RetryResult(entryId, identity.eventId(), identity.payloadHash(), request.requestId(), duplicate);
|
||||
}
|
||||
private static void validateId(String entryId) {
|
||||
if (entryId == null || !entryId.matches("[A-Za-z0-9-]{1,64}")) throw new BizException(ResultCode.BAD_REQUEST, "事件记录标识无效");
|
||||
}
|
||||
private static void validateRequest(RetryRequest request) {
|
||||
if (request == null) throw new BizException(ResultCode.BAD_REQUEST, "重试参数不能为空");
|
||||
try {
|
||||
if (!UUID.fromString(request.requestId()).toString().equalsIgnoreCase(request.requestId())) throw new IllegalArgumentException();
|
||||
} catch (RuntimeException failure) { throw new BizException(ResultCode.BAD_REQUEST, "requestId 必须为标准 UUID"); }
|
||||
requireText(request.operator(), 128, "操作人");
|
||||
requireText(request.reason(), 500, "原因");
|
||||
if (request.expectedPayloadHash() == null || !request.expectedPayloadHash().matches("sha256:[0-9a-fA-F]{64}")) {
|
||||
throw new BizException(ResultCode.BAD_REQUEST, "必须提供预期事件摘要");
|
||||
}
|
||||
}
|
||||
private static void requireText(String value, int length, String label) {
|
||||
if (value == null || value.isBlank() || value.length() > length || value.chars().anyMatch(Character::isISOControl)) {
|
||||
throw new BizException(ResultCode.BAD_REQUEST, label + "无效");
|
||||
}
|
||||
}
|
||||
private static BizException conflict(String message) { return new BizException(ResultCode.CONFLICT, message); }
|
||||
}
|
||||
@@ -0,0 +1,83 @@
|
||||
package com.baiyue.common.core;
|
||||
|
||||
import com.baiyue.common.core.outbox.EventDeliveryJournal;
|
||||
import com.baiyue.common.core.outbox.EventDeliveryJournal.Outcome;
|
||||
import org.junit.jupiter.api.BeforeEach;
|
||||
import org.junit.jupiter.api.Test;
|
||||
import org.springframework.jdbc.core.JdbcTemplate;
|
||||
import org.springframework.jdbc.datasource.DriverManagerDataSource;
|
||||
import org.springframework.jdbc.datasource.DataSourceTransactionManager;
|
||||
import org.springframework.transaction.support.TransactionTemplate;
|
||||
import java.time.LocalDateTime;
|
||||
import java.util.UUID;
|
||||
import static org.assertj.core.api.Assertions.*;
|
||||
|
||||
class EventDeliveryJournalTest {
|
||||
JdbcTemplate jdbc;
|
||||
EventDeliveryJournal journal;
|
||||
DataSourceTransactionManager transactions;
|
||||
@BeforeEach void setup() {
|
||||
var source = new DriverManagerDataSource("jdbc:h2:mem:" + UUID.randomUUID() + ";MODE=MySQL;DB_CLOSE_DELAY=-1", "sa", "");
|
||||
jdbc = new JdbcTemplate(source);
|
||||
transactions = new DataSourceTransactionManager(source);
|
||||
journal = new EventDeliveryJournal(jdbc, transactions);
|
||||
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 work(id INT PRIMARY KEY,state VARCHAR(20),owner VARCHAR(36))");
|
||||
jdbc.update("INSERT INTO work VALUES(1,'PENDING',NULL)");
|
||||
}
|
||||
boolean begin(String owner) {
|
||||
return journal.begin(owner, "firstleg", "FirstLegLoadingPlanCompleted", "1", UUID.randomUUID().toString(), "lastleg-events", LocalDateTime.now(),
|
||||
() -> jdbc.update("UPDATE work SET state='PROCESSING',owner=? WHERE id=1 AND state='PENDING'", owner) == 1);
|
||||
}
|
||||
int finish(String owner) { return jdbc.update("UPDATE work SET state='PUBLISHED',owner=NULL WHERE id=1 AND owner=?", owner); }
|
||||
String state() { return jdbc.queryForObject("SELECT state FROM work WHERE id=1", String.class); }
|
||||
String outcome(String attempt) { return jdbc.queryForObject("SELECT outcome FROM event_delivery_attempt WHERE attempt_id=?", String.class, attempt); }
|
||||
|
||||
@Test void startedIsCommittedBeforeCallerContinuesEvenWithinAnOuterTransaction() {
|
||||
new TransactionTemplate(transactions).executeWithoutResult(status -> {
|
||||
assertThat(begin("a")).isTrue();
|
||||
// Another connection sees STARTED before any HTTP would be invoked.
|
||||
var other = new JdbcTemplate(jdbc.getDataSource());
|
||||
assertThat(other.queryForObject("SELECT outcome FROM event_delivery_attempt", String.class)).isEqualTo("STARTED");
|
||||
status.setRollbackOnly();
|
||||
});
|
||||
assertThat(state()).isEqualTo("PROCESSING");
|
||||
assertThat(outcome("a")).isEqualTo("STARTED"); // process interruption leaves diagnostic evidence
|
||||
assertThat(begin("b")).isFalse();
|
||||
assertThat(jdbc.queryForObject("SELECT COUNT(*) FROM event_delivery_attempt", Integer.class)).isEqualTo(1);
|
||||
}
|
||||
@Test void startAuditFailureRollsBackTheClaim() {
|
||||
assertThat(begin("a")).isTrue();
|
||||
jdbc.update("UPDATE work SET state='PENDING',owner=NULL");
|
||||
assertThatThrownBy(() -> begin("a")).isInstanceOf(RuntimeException.class);
|
||||
assertThat(state()).isEqualTo("PENDING");
|
||||
}
|
||||
@Test void completionAuditFailureRollsBackOutboxAndPreservesStartedRecord() {
|
||||
begin("a");
|
||||
jdbc.execute("ALTER TABLE event_delivery_attempt ADD CONSTRAINT reject_applied CHECK(outcome <> 'APPLIED')");
|
||||
assertThatThrownBy(() -> journal.complete("a", Outcome.APPLIED, LocalDateTime.now(), null, null, () -> finish("a"))).isInstanceOf(RuntimeException.class);
|
||||
assertThat(state()).isEqualTo("PROCESSING");
|
||||
assertThat(outcome("a")).isEqualTo("STARTED");
|
||||
jdbc.execute("ALTER TABLE event_delivery_attempt DROP CONSTRAINT reject_applied");
|
||||
assertThat(journal.complete("a", Outcome.APPLIED, LocalDateTime.now(), null, null, () -> finish("a"))).isTrue();
|
||||
assertThat(state()).isEqualTo("PUBLISHED");
|
||||
assertThat(outcome("a")).isEqualTo("APPLIED");
|
||||
assertThat(jdbc.queryForObject("SELECT duration_ms FROM event_delivery_attempt", Long.class)).isNotNegative();
|
||||
}
|
||||
@Test void recoveryAndLateResultCannotOverwriteTheNewOwnerOrItsAudit() {
|
||||
begin("a");
|
||||
var retryAt = LocalDateTime.now().plusSeconds(5);
|
||||
assertThat(journal.complete("a", Outcome.LEASE_EXPIRED, LocalDateTime.now(), "LEASE_EXPIRED", retryAt,
|
||||
() -> jdbc.update("UPDATE work SET state='PENDING',owner=NULL WHERE owner='a'"))).isTrue();
|
||||
begin("b");
|
||||
assertThat(journal.complete("a", Outcome.APPLIED, LocalDateTime.now(), null, null, () -> finish("a"))).isFalse();
|
||||
assertThat(outcome("a")).isEqualTo("LEASE_EXPIRED");
|
||||
assertThat(outcome("b")).isEqualTo("STARTED");
|
||||
assertThat(state()).isEqualTo("PROCESSING");
|
||||
}
|
||||
@Test void unexpectedCompletionWithoutStartedRecordRollsBackInsteadOfUsingAFallback() {
|
||||
jdbc.update("UPDATE work SET state='PROCESSING',owner='missing'");
|
||||
assertThatThrownBy(() -> journal.complete("missing", Outcome.APPLIED, LocalDateTime.now(), null, null, () -> finish("missing"))).isInstanceOf(RuntimeException.class);
|
||||
assertThat(state()).isEqualTo("PROCESSING");
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,23 @@
|
||||
package com.baiyue.common.core;
|
||||
|
||||
import com.baiyue.common.core.exception.BizException;
|
||||
import com.baiyue.common.core.outbox.EventOperationsTokenGuard;
|
||||
import org.junit.jupiter.api.Test;
|
||||
import static org.assertj.core.api.Assertions.*;
|
||||
|
||||
class EventOperationsTokenGuardTest {
|
||||
@Test void missingOrShortConfigurationDisablesOperations() {
|
||||
for (String configured : new String[] {null, "", "short"}) {
|
||||
var guard = new EventOperationsTokenGuard(configured);
|
||||
assertThatThrownBy(() -> guard.require(configured)).isInstanceOf(BizException.class);
|
||||
}
|
||||
}
|
||||
@Test void onlyExactCredentialIsAccepted() {
|
||||
String token = "operation-secret-".repeat(3);
|
||||
var guard = new EventOperationsTokenGuard(token);
|
||||
assertThatCode(() -> guard.require(token)).doesNotThrowAnyException();
|
||||
for (String supplied : new String[] {null, "", token + " ", token.toUpperCase(), "a".repeat(4097)}) {
|
||||
assertThatThrownBy(() -> guard.require(supplied)).isInstanceOf(BizException.class);
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,88 @@
|
||||
package com.baiyue.common.core;
|
||||
|
||||
import com.baiyue.common.core.exception.BizException;
|
||||
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.springframework.jdbc.core.JdbcTemplate;
|
||||
import org.springframework.jdbc.datasource.DataSourceTransactionManager;
|
||||
import java.util.Set;
|
||||
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.*;
|
||||
|
||||
class OutboxOperationsTest {
|
||||
private static final String HASH = "sha256:" + "a".repeat(64);
|
||||
private JdbcTemplate jdbc;
|
||||
private OutboxOperations operations;
|
||||
|
||||
@BeforeEach void setup() {
|
||||
var source = new JdbcDataSource();
|
||||
source.setURL("jdbc:h2:mem:" + UUID.randomUUID() + ";MODE=MySQL;DB_CLOSE_DELAY=-1;LOCK_TIMEOUT=5000");
|
||||
jdbc = new JdbcTemplate(source);
|
||||
jdbc.execute("CREATE TABLE events(id VARCHAR(64) PRIMARY KEY,payload_json VARCHAR(200),status VARCHAR(20),retry_count INT,last_error VARCHAR(1000),next_retry_at TIMESTAMP,locked_by VARCHAR(100),locked_at TIMESTAMP)");
|
||||
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)");
|
||||
operations = new OutboxOperations(jdbc, new DataSourceTransactionManager(source),
|
||||
new OutboxOperations.Layout("events", "id", "status", Set.of("business.completed")),
|
||||
payload -> new OutboxOperations.Identity(payload, "business.completed", HASH));
|
||||
jdbc.update("INSERT INTO events VALUES('1','event-1','DEAD',12,'timeout',CURRENT_TIMESTAMP,'old-worker',CURRENT_TIMESTAMP)");
|
||||
}
|
||||
private OutboxOperations.RetryRequest request() {
|
||||
return new OutboxOperations.RetryRequest(UUID.randomUUID().toString(), "operator-7", "dependency restored", HASH);
|
||||
}
|
||||
|
||||
@Test void retryPreservesPayloadAndAuditsPreviousFailureThenDeduplicates() {
|
||||
var request = request();
|
||||
assertThat(operations.retry("1", request).duplicate()).isFalse();
|
||||
assertThat(operations.retry("1", request).duplicate()).isTrue();
|
||||
assertThat(jdbc.queryForMap("SELECT * FROM events WHERE id='1'"))
|
||||
.containsEntry("PAYLOAD_JSON", "event-1").containsEntry("STATUS", "PENDING")
|
||||
.containsEntry("RETRY_COUNT", 0).containsEntry("LOCKED_BY", null).containsEntry("LAST_ERROR", null);
|
||||
assertThat(jdbc.queryForMap("SELECT * FROM event_delivery_retry_audit"))
|
||||
.containsEntry("PREVIOUS_FAILURES", 12).containsEntry("PREVIOUS_ERROR", "timeout")
|
||||
.containsEntry("OPERATOR_NAME", "operator-7").containsEntry("PAYLOAD_HASH", HASH);
|
||||
}
|
||||
|
||||
@Test void auditInsertFailureRollsBackRequeue() {
|
||||
jdbc.execute("ALTER TABLE event_delivery_retry_audit ADD CONSTRAINT reject_audit CHECK(previous_failures < 0)");
|
||||
assertThatThrownBy(() -> operations.retry("1", request())).isInstanceOf(org.springframework.dao.DataIntegrityViolationException.class);
|
||||
assertThat(operations.inspect("1").status()).isEqualTo("DEAD");
|
||||
assertThat(operations.inspect("1").failures()).isEqualTo(12);
|
||||
assertThat(jdbc.queryForObject("SELECT COUNT(*) FROM event_delivery_retry_audit", Integer.class)).isZero();
|
||||
}
|
||||
|
||||
@Test void staleHashAndNonDeadAndReusedRequestAreRejected() {
|
||||
var original = request();
|
||||
var stale = new OutboxOperations.RetryRequest(original.requestId(), original.operator(), original.reason(), "sha256:" + "b".repeat(64));
|
||||
assertThatThrownBy(() -> operations.retry("1", stale)).isInstanceOf(BizException.class);
|
||||
operations.retry("1", original);
|
||||
assertThatThrownBy(() -> operations.retry("1", request())).isInstanceOf(BizException.class);
|
||||
var changed = new OutboxOperations.RetryRequest(original.requestId(), "someone-else", original.reason(), HASH);
|
||||
assertThatThrownBy(() -> operations.retry("1", changed)).isInstanceOf(BizException.class);
|
||||
assertThat(jdbc.queryForObject("SELECT COUNT(*) FROM event_delivery_retry_audit", Integer.class)).isEqualTo(1);
|
||||
}
|
||||
|
||||
@Test void concurrentSameRequestHasOneAuditAndOneDuplicate() throws Exception {
|
||||
var request = request();
|
||||
var start = new CountDownLatch(1);
|
||||
var pool = Executors.newFixedThreadPool(2);
|
||||
try {
|
||||
var first = pool.submit(() -> { start.await(); return operations.retry("1", request); });
|
||||
var second = pool.submit(() -> { start.await(); return operations.retry("1", request); });
|
||||
start.countDown();
|
||||
assertThat(first.get(10, TimeUnit.SECONDS).duplicate()).isNotEqualTo(second.get(10, TimeUnit.SECONDS).duplicate());
|
||||
assertThat(jdbc.queryForObject("SELECT COUNT(*) FROM event_delivery_retry_audit", Integer.class)).isEqualTo(1);
|
||||
} finally { pool.shutdownNow(); }
|
||||
}
|
||||
|
||||
@Test void crossEntryRequestCollisionRollsBackSecondEntry() {
|
||||
jdbc.update("INSERT INTO events VALUES('2','event-2','DEAD',12,'timeout',CURRENT_TIMESTAMP,NULL,NULL)");
|
||||
var request = request();
|
||||
operations.retry("1", request);
|
||||
assertThatThrownBy(() -> operations.retry("2", request)).isInstanceOf(BizException.class);
|
||||
assertThat(operations.inspect("2").status()).isEqualTo("DEAD");
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user