merge: integrate JPA production capability
# Conflicts: # .github/ci-gate-matrix.yml # .github/scripts/verify-gate-matrix.sh # .github/workflows/ci-quality-gates.yml # src/adapter/outbound/persistence-jpa/src/main/java/dev/caskeleton/adapter/outbound/persistence/transaction/SpringTransactionPort.java
This commit is contained in:
@@ -296,6 +296,21 @@ application 계층이 in-flight 대기·replay **정책** 을 소유하고, 저
|
||||
- retryable: mismatch / in-flight 모두 `false`. 클라이언트는 body 를 고치거나 결과를 polling 해야지
|
||||
단순 재시도를 하면 안 된다.
|
||||
|
||||
### Owner-safe idempotency V2
|
||||
|
||||
`idempotency.v2`는 V1의 scope-only `tryBegin/complete/discard`를 대체하는 additive contract다.
|
||||
provider가 JPA인지 Redis인지와 무관하게 claim에는 secure owner token과 stable operation ID가
|
||||
필요하고, 모든 mutation은 owner/attempt/claim-operation/state-revision을 검증한다.
|
||||
|
||||
- processing lease와 completed replay TTL을 분리한다.
|
||||
- expired `CLAIMED`만 takeover하고 expired `EXECUTING`은 `RECOVERY_REQUIRED`로 닫는다.
|
||||
- complete/fail/release는 operation ID와 result digest가 같은 재호출만 prior result로 replay한다.
|
||||
- `SAME_STORE_TRANSACTIONAL` JPA profile의 response는 8 KiB 이하 inline 값만 지원한다.
|
||||
- raw principal/client key는 versioned HMAC scope digest로 바꾼 뒤 adapter에 전달한다.
|
||||
|
||||
V1은 rolling migration compatibility를 위해 유지된다. 새 reliability profile이 V2 claim과 V1
|
||||
scope-only mutation을 섞는 것은 금지한다.
|
||||
|
||||
---
|
||||
|
||||
## 트랜잭셔널 아웃박스 릴레이 (outbox)
|
||||
@@ -423,6 +438,26 @@ claim → 트랜잭션 밖에서 발행 → at-least-once 보장.
|
||||
adapter에 전달하는 framework-free outbound contract. 구조화 ERROR 필드와 runbook 렌더링은
|
||||
messaging adapter가 소유한다.
|
||||
|
||||
### Immutable outbox/polling delivery V2
|
||||
|
||||
`outbox.v2`는 domain event intent와 delivery state를 분리한다.
|
||||
|
||||
- `NewOutboxEventV2`의 aggregate version과 deterministic ordinal이 ordering authority다.
|
||||
- `OutboxAppendPortV2`는 caller의 primary write transaction에 참여하고 publication epoch와
|
||||
authority를 DB control row에서 얻는다.
|
||||
- immutable event ID 충돌과 aggregate ordering tuple 충돌은 서로 다른 outcome이다.
|
||||
- `OutboxPollingDeliveryPortV2`는 publish 밖의 짧은 claim/completion transaction만 소유하고
|
||||
owner/token/attempt/version/epoch CAS로 stale relay를 거절한다.
|
||||
- broker publish와 DB completion 사이 ACK 유실은 stable event ID의 duplicate publish를 만들 수
|
||||
있으므로 exactly-once delivery로 표현하지 않는다.
|
||||
|
||||
### Same-store inbox
|
||||
|
||||
`inbox.InboxStorePort`는 broker redelivery를 DB business mutation과 같은 transaction에서
|
||||
deduplicate한다. `RECEIVED -> PROCESSING -> COMPLETED`가 기본이며 expired `PROCESSING`은 blind
|
||||
takeover하지 않는다. broker ACK는 transaction commit 이후 adapter 바깥에서만 수행하고 remote
|
||||
side effect는 outbox/workflow로 옮긴다.
|
||||
|
||||
---
|
||||
|
||||
## 분산 락 (lock)
|
||||
|
||||
+20
@@ -0,0 +1,20 @@
|
||||
package dev.caskeleton.application.idempotency.v2;
|
||||
|
||||
import dev.caskeleton.application.transaction.OperationId;
|
||||
import java.util.Objects;
|
||||
import java.util.regex.Pattern;
|
||||
|
||||
/** Caller-retained identity for one claim send/retry sequence. */
|
||||
public record IdempotencyClaimAttempt(String ownerToken, OperationId operationId) {
|
||||
|
||||
private static final Pattern OWNER_TOKEN = Pattern.compile("[0-9a-f]{64}");
|
||||
|
||||
public IdempotencyClaimAttempt {
|
||||
Objects.requireNonNull(ownerToken, "ownerToken");
|
||||
Objects.requireNonNull(operationId, "operationId");
|
||||
if (!OWNER_TOKEN.matcher(ownerToken).matches()) {
|
||||
throw new IllegalArgumentException(
|
||||
"owner token must be a 64-character lowercase hexadecimal secure-random value");
|
||||
}
|
||||
}
|
||||
}
|
||||
+73
@@ -0,0 +1,73 @@
|
||||
package dev.caskeleton.application.idempotency.v2;
|
||||
|
||||
import dev.caskeleton.application.idempotency.StoredResponse;
|
||||
import dev.caskeleton.application.transaction.OperationId;
|
||||
import java.time.Duration;
|
||||
import java.time.Instant;
|
||||
import java.util.Objects;
|
||||
|
||||
/** Exhaustive result of an atomic owner-safe claim. */
|
||||
public sealed interface IdempotencyClaimOutcome {
|
||||
|
||||
record Acquired(IdempotencyOwner owner, Instant processingLeaseUntil)
|
||||
implements IdempotencyClaimOutcome {
|
||||
public Acquired {
|
||||
Objects.requireNonNull(owner, "owner");
|
||||
Objects.requireNonNull(processingLeaseUntil, "processingLeaseUntil");
|
||||
}
|
||||
}
|
||||
|
||||
record ReplayedAcquire(IdempotencyOwner owner, Instant processingLeaseUntil)
|
||||
implements IdempotencyClaimOutcome {
|
||||
public ReplayedAcquire {
|
||||
Objects.requireNonNull(owner, "owner");
|
||||
Objects.requireNonNull(processingLeaseUntil, "processingLeaseUntil");
|
||||
}
|
||||
}
|
||||
|
||||
record TakenOverClaimed(IdempotencyOwner owner, Instant processingLeaseUntil)
|
||||
implements IdempotencyClaimOutcome {
|
||||
public TakenOverClaimed {
|
||||
Objects.requireNonNull(owner, "owner");
|
||||
Objects.requireNonNull(processingLeaseUntil, "processingLeaseUntil");
|
||||
}
|
||||
}
|
||||
|
||||
record CompletedReplay(StoredResponse response, Instant replayUntil)
|
||||
implements IdempotencyClaimOutcome {
|
||||
public CompletedReplay {
|
||||
Objects.requireNonNull(response, "response");
|
||||
Objects.requireNonNull(replayUntil, "replayUntil");
|
||||
}
|
||||
}
|
||||
|
||||
record InProgress(Duration retryAfter, long currentAttempt) implements IdempotencyClaimOutcome {
|
||||
public InProgress {
|
||||
Objects.requireNonNull(retryAfter, "retryAfter");
|
||||
if (retryAfter.isNegative() || currentAttempt < 1) {
|
||||
throw new IllegalArgumentException(
|
||||
"retry-after must be non-negative and current attempt must be positive");
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
record RecoveryRequired(long currentAttempt) implements IdempotencyClaimOutcome {
|
||||
public RecoveryRequired {
|
||||
if (currentAttempt < 1) {
|
||||
throw new IllegalArgumentException("current attempt must be positive");
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
record FingerprintMismatch() implements IdempotencyClaimOutcome {}
|
||||
|
||||
record OwnerOperationConflict() implements IdempotencyClaimOutcome {}
|
||||
|
||||
record Indeterminate(OperationId operationId) implements IdempotencyClaimOutcome {
|
||||
public Indeterminate {
|
||||
Objects.requireNonNull(operationId, "operationId");
|
||||
}
|
||||
}
|
||||
|
||||
record Unavailable() implements IdempotencyClaimOutcome {}
|
||||
}
|
||||
+41
@@ -0,0 +1,41 @@
|
||||
package dev.caskeleton.application.idempotency.v2;
|
||||
|
||||
import dev.caskeleton.application.idempotency.RequestFingerprint;
|
||||
import java.time.Duration;
|
||||
import java.util.Objects;
|
||||
|
||||
/** Complete provider-neutral intent for one atomic owner-safe claim. */
|
||||
public record IdempotencyClaimRequest(
|
||||
IdempotencyScopeDigest scope,
|
||||
RequestFingerprint requestFingerprint,
|
||||
IdempotencyClaimAttempt claimAttempt,
|
||||
Duration processingLeaseTtl,
|
||||
Duration replayTtl,
|
||||
String responseCodecId,
|
||||
int policyRevision) {
|
||||
|
||||
private static final Duration MAXIMUM_PROCESSING_LEASE = Duration.ofHours(1);
|
||||
private static final Duration MAXIMUM_REPLAY_TTL = Duration.ofDays(30);
|
||||
|
||||
public IdempotencyClaimRequest {
|
||||
Objects.requireNonNull(scope, "scope");
|
||||
Objects.requireNonNull(requestFingerprint, "requestFingerprint");
|
||||
Objects.requireNonNull(claimAttempt, "claimAttempt");
|
||||
requirePositiveBounded("processing lease TTL", processingLeaseTtl, MAXIMUM_PROCESSING_LEASE);
|
||||
requirePositiveBounded("replay TTL", replayTtl, MAXIMUM_REPLAY_TTL);
|
||||
Objects.requireNonNull(responseCodecId, "responseCodecId");
|
||||
if (responseCodecId.isBlank() || responseCodecId.length() > 64) {
|
||||
throw new IllegalArgumentException("response codec ID must contain 1-64 characters");
|
||||
}
|
||||
if (policyRevision < 1) {
|
||||
throw new IllegalArgumentException("policy revision must be positive");
|
||||
}
|
||||
}
|
||||
|
||||
private static void requirePositiveBounded(String name, Duration value, Duration maximum) {
|
||||
Objects.requireNonNull(value, name);
|
||||
if (value.isZero() || value.isNegative() || value.compareTo(maximum) > 0) {
|
||||
throw new IllegalArgumentException(name + " must be positive and at most " + maximum);
|
||||
}
|
||||
}
|
||||
}
|
||||
+14
@@ -0,0 +1,14 @@
|
||||
package dev.caskeleton.application.idempotency.v2;
|
||||
|
||||
/** Outcomes of an owner-safe EXECUTING to COMPLETED transition. */
|
||||
public enum IdempotencyCompleteOutcome {
|
||||
COMPLETED,
|
||||
ALREADY_COMPLETED_SAME_RESULT,
|
||||
RESPONSE_CONFLICT,
|
||||
ABSENT,
|
||||
NOT_OWNER,
|
||||
NOT_IN_PROGRESS,
|
||||
OPERATION_CONFLICT,
|
||||
INDETERMINATE,
|
||||
UNAVAILABLE
|
||||
}
|
||||
+14
@@ -0,0 +1,14 @@
|
||||
package dev.caskeleton.application.idempotency.v2;
|
||||
|
||||
/** Outcomes of an owner-safe failure transition. */
|
||||
public enum IdempotencyFailOutcome {
|
||||
MARKED_RETRYABLE,
|
||||
MARKED_ABANDONED,
|
||||
ALREADY_MARKED_SAME_OPERATION,
|
||||
ABSENT,
|
||||
NOT_OWNER,
|
||||
NOT_IN_PROGRESS,
|
||||
OPERATION_CONFLICT,
|
||||
INDETERMINATE,
|
||||
UNAVAILABLE
|
||||
}
|
||||
+7
@@ -0,0 +1,7 @@
|
||||
package dev.caskeleton.application.idempotency.v2;
|
||||
|
||||
/** Whether a failed action is proven retryable or requires explicit effect reconciliation. */
|
||||
public enum IdempotencyFailureDisposition {
|
||||
NO_EFFECT_RETRYABLE,
|
||||
EFFECT_UNKNOWN_ABANDONED
|
||||
}
|
||||
+28
@@ -0,0 +1,28 @@
|
||||
package dev.caskeleton.application.idempotency.v2;
|
||||
|
||||
import dev.caskeleton.application.idempotency.StoredResponse;
|
||||
import java.time.Instant;
|
||||
import java.util.Objects;
|
||||
import java.util.Optional;
|
||||
|
||||
/** Inspection classification with only the data meaningful for that classification. */
|
||||
public record IdempotencyInspection(
|
||||
IdempotencyInspectionOutcome outcome,
|
||||
Optional<IdempotencyOwner> owner,
|
||||
Optional<Instant> processingLeaseUntil,
|
||||
Optional<StoredResponse> response,
|
||||
Optional<Instant> replayUntil) {
|
||||
|
||||
public IdempotencyInspection {
|
||||
Objects.requireNonNull(outcome, "outcome");
|
||||
Objects.requireNonNull(owner, "owner");
|
||||
Objects.requireNonNull(processingLeaseUntil, "processingLeaseUntil");
|
||||
Objects.requireNonNull(response, "response");
|
||||
Objects.requireNonNull(replayUntil, "replayUntil");
|
||||
}
|
||||
|
||||
public static IdempotencyInspection outcome(IdempotencyInspectionOutcome outcome) {
|
||||
return new IdempotencyInspection(
|
||||
outcome, Optional.empty(), Optional.empty(), Optional.empty(), Optional.empty());
|
||||
}
|
||||
}
|
||||
+15
@@ -0,0 +1,15 @@
|
||||
package dev.caskeleton.application.idempotency.v2;
|
||||
|
||||
/** Recovery-safe classifications returned by {@link IdempotencyStorePortV2#inspect}. */
|
||||
public enum IdempotencyInspectionOutcome {
|
||||
ABSENT,
|
||||
CLAIMED_SAME_OPERATION,
|
||||
EXECUTING_SAME_OPERATION,
|
||||
COMPLETED_REPLAY,
|
||||
IN_PROGRESS_OTHER,
|
||||
FAILED_RETRYABLE,
|
||||
ABANDONED,
|
||||
FINGERPRINT_MISMATCH,
|
||||
OPERATION_CONFLICT,
|
||||
UNAVAILABLE
|
||||
}
|
||||
+17
@@ -0,0 +1,17 @@
|
||||
package dev.caskeleton.application.idempotency.v2;
|
||||
|
||||
import dev.caskeleton.application.idempotency.RequestFingerprint;
|
||||
import java.util.Objects;
|
||||
|
||||
/** Read-only recovery request after a claim or transition response was lost. */
|
||||
public record IdempotencyInspectionRequest(
|
||||
IdempotencyScopeDigest scope,
|
||||
RequestFingerprint requestFingerprint,
|
||||
IdempotencyClaimAttempt claimAttempt) {
|
||||
|
||||
public IdempotencyInspectionRequest {
|
||||
Objects.requireNonNull(scope, "scope");
|
||||
Objects.requireNonNull(requestFingerprint, "requestFingerprint");
|
||||
Objects.requireNonNull(claimAttempt, "claimAttempt");
|
||||
}
|
||||
}
|
||||
+35
@@ -0,0 +1,35 @@
|
||||
package dev.caskeleton.application.idempotency.v2;
|
||||
|
||||
import java.util.Objects;
|
||||
import java.util.Optional;
|
||||
import java.util.function.Predicate;
|
||||
|
||||
/**
|
||||
* Operation-specific typed outcome plus a replacement owner handle when ownership remains valid.
|
||||
*/
|
||||
public final class IdempotencyMutationResult<O extends Enum<O>> {
|
||||
|
||||
private final O outcome;
|
||||
private final IdempotencyOwner owner;
|
||||
|
||||
public IdempotencyMutationResult(O outcome, IdempotencyOwner owner, Predicate<O> carriesOwner) {
|
||||
this.outcome = Objects.requireNonNull(outcome, "outcome");
|
||||
Objects.requireNonNull(carriesOwner, "carriesOwner");
|
||||
boolean expectedOwner = carriesOwner.test(outcome);
|
||||
if (expectedOwner && owner == null) {
|
||||
throw new IllegalArgumentException(outcome + " must carry the current owner handle");
|
||||
}
|
||||
if (!expectedOwner && owner != null) {
|
||||
throw new IllegalArgumentException(outcome + " must not carry an owner handle");
|
||||
}
|
||||
this.owner = owner;
|
||||
}
|
||||
|
||||
public O outcome() {
|
||||
return outcome;
|
||||
}
|
||||
|
||||
public Optional<IdempotencyOwner> owner() {
|
||||
return Optional.ofNullable(owner);
|
||||
}
|
||||
}
|
||||
+37
@@ -0,0 +1,37 @@
|
||||
package dev.caskeleton.application.idempotency.v2;
|
||||
|
||||
import dev.caskeleton.application.transaction.OperationId;
|
||||
import java.util.Objects;
|
||||
|
||||
/**
|
||||
* Owner handle carrying the full optimistic CAS tuple.
|
||||
*
|
||||
* <p>Every successful state-changing operation returns a replacement handle with the incremented
|
||||
* state revision. A stale handle is never silently accepted.
|
||||
*/
|
||||
public record IdempotencyOwner(
|
||||
IdempotencyScopeDigest scope,
|
||||
String ownerToken,
|
||||
long attempt,
|
||||
long stateRevision,
|
||||
OperationId claimOperationId) {
|
||||
|
||||
public IdempotencyOwner {
|
||||
Objects.requireNonNull(scope, "scope");
|
||||
Objects.requireNonNull(ownerToken, "ownerToken");
|
||||
Objects.requireNonNull(claimOperationId, "claimOperationId");
|
||||
if (ownerToken.isBlank() || ownerToken.length() > 128) {
|
||||
throw new IllegalArgumentException("owner token must contain 1-128 characters");
|
||||
}
|
||||
if (attempt < 1) {
|
||||
throw new IllegalArgumentException("attempt must be positive");
|
||||
}
|
||||
if (stateRevision < 0) {
|
||||
throw new IllegalArgumentException("state revision must be non-negative");
|
||||
}
|
||||
}
|
||||
|
||||
public IdempotencyOwner withStateRevision(long nextRevision) {
|
||||
return new IdempotencyOwner(scope, ownerToken, attempt, nextRevision, claimOperationId);
|
||||
}
|
||||
}
|
||||
+13
@@ -0,0 +1,13 @@
|
||||
package dev.caskeleton.application.idempotency.v2;
|
||||
|
||||
/** Outcomes of releasing a claim only while business execution has not started. */
|
||||
public enum IdempotencyReleaseOutcome {
|
||||
RELEASED_BEFORE_EXECUTION,
|
||||
ALREADY_RELEASED_SAME_OPERATION,
|
||||
ABSENT,
|
||||
NOT_OWNER,
|
||||
EXECUTION_ALREADY_STARTED,
|
||||
OPERATION_CONFLICT,
|
||||
INDETERMINATE,
|
||||
UNAVAILABLE
|
||||
}
|
||||
+23
@@ -0,0 +1,23 @@
|
||||
package dev.caskeleton.application.idempotency.v2;
|
||||
|
||||
/** Outcomes of an owner-safe processing lease renewal. */
|
||||
public enum IdempotencyRenewOutcome {
|
||||
RENEWED(true),
|
||||
ALREADY_RENEWED_SAME_OPERATION(true),
|
||||
ABSENT(false),
|
||||
NOT_OWNER(false),
|
||||
NOT_IN_PROGRESS(false),
|
||||
OPERATION_CONFLICT(false),
|
||||
INDETERMINATE(false),
|
||||
UNAVAILABLE(false);
|
||||
|
||||
private final boolean carriesOwner;
|
||||
|
||||
IdempotencyRenewOutcome(boolean carriesOwner) {
|
||||
this.carriesOwner = carriesOwner;
|
||||
}
|
||||
|
||||
public boolean carriesOwner() {
|
||||
return carriesOwner;
|
||||
}
|
||||
}
|
||||
+32
@@ -0,0 +1,32 @@
|
||||
package dev.caskeleton.application.idempotency.v2;
|
||||
|
||||
import java.util.Objects;
|
||||
import java.util.regex.Pattern;
|
||||
|
||||
/**
|
||||
* Provider-neutral, versioned HMAC digest of the canonical idempotency scope.
|
||||
*
|
||||
* <p>The raw client key and principal must be digested before this value is constructed. Database,
|
||||
* Redis, logs, and metrics receive only this opaque value.
|
||||
*/
|
||||
public record IdempotencyScopeDigest(String digest, int keyDigestVersion, String operationCode) {
|
||||
|
||||
private static final Pattern LOWERCASE_SHA_256 = Pattern.compile("[0-9a-f]{64}");
|
||||
private static final Pattern OPERATION_CODE = Pattern.compile("[A-Z][A-Z0-9_]{0,63}");
|
||||
|
||||
public IdempotencyScopeDigest {
|
||||
Objects.requireNonNull(digest, "digest");
|
||||
Objects.requireNonNull(operationCode, "operationCode");
|
||||
if (!LOWERCASE_SHA_256.matcher(digest).matches()) {
|
||||
throw new IllegalArgumentException(
|
||||
"scope digest must be a 64-character lowercase hexadecimal HMAC-SHA-256 value");
|
||||
}
|
||||
if (keyDigestVersion < 1) {
|
||||
throw new IllegalArgumentException("key digest version must be positive");
|
||||
}
|
||||
if (!OPERATION_CODE.matcher(operationCode).matches()) {
|
||||
throw new IllegalArgumentException(
|
||||
"operation code must be 1-64 uppercase ASCII letters, digits, or underscores");
|
||||
}
|
||||
}
|
||||
}
|
||||
+23
@@ -0,0 +1,23 @@
|
||||
package dev.caskeleton.application.idempotency.v2;
|
||||
|
||||
/** Outcomes of the CLAIMED to EXECUTING owner-safe transition. */
|
||||
public enum IdempotencyStartOutcome {
|
||||
STARTED(true),
|
||||
ALREADY_STARTED_SAME_OPERATION(true),
|
||||
ABSENT(false),
|
||||
NOT_OWNER(false),
|
||||
NOT_CLAIMED(false),
|
||||
OPERATION_CONFLICT(false),
|
||||
INDETERMINATE(false),
|
||||
UNAVAILABLE(false);
|
||||
|
||||
private final boolean carriesOwner;
|
||||
|
||||
IdempotencyStartOutcome(boolean carriesOwner) {
|
||||
this.carriesOwner = carriesOwner;
|
||||
}
|
||||
|
||||
public boolean carriesOwner() {
|
||||
return carriesOwner;
|
||||
}
|
||||
}
|
||||
+10
@@ -0,0 +1,10 @@
|
||||
package dev.caskeleton.application.idempotency.v2;
|
||||
|
||||
/** Owner-safe idempotency V2 state machine. */
|
||||
public enum IdempotencyState {
|
||||
CLAIMED,
|
||||
EXECUTING,
|
||||
COMPLETED,
|
||||
FAILED_RETRYABLE,
|
||||
ABANDONED
|
||||
}
|
||||
+37
@@ -0,0 +1,37 @@
|
||||
package dev.caskeleton.application.idempotency.v2;
|
||||
|
||||
import dev.caskeleton.application.idempotency.StoredResponse;
|
||||
import dev.caskeleton.application.transaction.OperationId;
|
||||
import java.time.Duration;
|
||||
|
||||
/**
|
||||
* Provider-neutral owner-safe idempotency state machine.
|
||||
*
|
||||
* <p>V1 remains source-compatible during migration, but new reliability profiles must use this
|
||||
* complete contract rather than combining V2 claim with scope-only V1 mutations.
|
||||
*/
|
||||
public interface IdempotencyStorePortV2 {
|
||||
|
||||
IdempotencyClaimAttempt newClaimAttempt(OperationId operationId);
|
||||
|
||||
IdempotencyClaimOutcome claim(IdempotencyClaimRequest request);
|
||||
|
||||
IdempotencyMutationResult<IdempotencyStartOutcome> markExecutionStarted(
|
||||
IdempotencyOwner owner, OperationId operationId);
|
||||
|
||||
IdempotencyMutationResult<IdempotencyRenewOutcome> renew(
|
||||
IdempotencyOwner owner, Duration processingLeaseTtl, OperationId operationId);
|
||||
|
||||
IdempotencyCompleteOutcome complete(
|
||||
IdempotencyOwner owner, StoredResponse response, Duration replayTtl, OperationId operationId);
|
||||
|
||||
IdempotencyFailOutcome markFailed(
|
||||
IdempotencyOwner owner,
|
||||
IdempotencyFailureDisposition disposition,
|
||||
Duration retention,
|
||||
OperationId operationId);
|
||||
|
||||
IdempotencyReleaseOutcome releaseBeforeExecution(IdempotencyOwner owner, OperationId operationId);
|
||||
|
||||
IdempotencyInspection inspect(IdempotencyInspectionRequest request);
|
||||
}
|
||||
+20
@@ -0,0 +1,20 @@
|
||||
package dev.caskeleton.application.inbox;
|
||||
|
||||
import dev.caskeleton.application.transaction.OperationId;
|
||||
import java.util.Objects;
|
||||
import java.util.regex.Pattern;
|
||||
|
||||
/** Caller-retained inbox claim identity used to replay a lost claim response. */
|
||||
public record InboxClaimAttempt(String ownerToken, OperationId operationId) {
|
||||
|
||||
private static final Pattern TOKEN = Pattern.compile("[0-9a-f]{64}");
|
||||
|
||||
public InboxClaimAttempt {
|
||||
Objects.requireNonNull(ownerToken, "ownerToken");
|
||||
Objects.requireNonNull(operationId, "operationId");
|
||||
if (!TOKEN.matcher(ownerToken).matches()) {
|
||||
throw new IllegalArgumentException(
|
||||
"owner token must be a 64-character lowercase hexadecimal value");
|
||||
}
|
||||
}
|
||||
}
|
||||
+40
@@ -0,0 +1,40 @@
|
||||
package dev.caskeleton.application.inbox;
|
||||
|
||||
import java.time.Duration;
|
||||
import java.time.Instant;
|
||||
import java.util.Objects;
|
||||
|
||||
/** Exhaustive atomic inbox claim classification. */
|
||||
public sealed interface InboxClaimOutcome {
|
||||
|
||||
record Acquired(InboxOwner owner, Instant leaseUntil) implements InboxClaimOutcome {
|
||||
public Acquired {
|
||||
Objects.requireNonNull(owner, "owner");
|
||||
Objects.requireNonNull(leaseUntil, "leaseUntil");
|
||||
}
|
||||
}
|
||||
|
||||
record ReplayedAcquire(InboxOwner owner, Instant leaseUntil) implements InboxClaimOutcome {
|
||||
public ReplayedAcquire {
|
||||
Objects.requireNonNull(owner, "owner");
|
||||
Objects.requireNonNull(leaseUntil, "leaseUntil");
|
||||
}
|
||||
}
|
||||
|
||||
record TakenOver(InboxOwner owner, Instant leaseUntil) implements InboxClaimOutcome {
|
||||
public TakenOver {
|
||||
Objects.requireNonNull(owner, "owner");
|
||||
Objects.requireNonNull(leaseUntil, "leaseUntil");
|
||||
}
|
||||
}
|
||||
|
||||
record Completed() implements InboxClaimOutcome {}
|
||||
|
||||
record InProgress(Duration retryAfter, long attempt) implements InboxClaimOutcome {}
|
||||
|
||||
record RecoveryRequired(long attempt) implements InboxClaimOutcome {}
|
||||
|
||||
record IntentMismatch() implements InboxClaimOutcome {}
|
||||
|
||||
record OwnerOperationConflict() implements InboxClaimOutcome {}
|
||||
}
|
||||
+35
@@ -0,0 +1,35 @@
|
||||
package dev.caskeleton.application.inbox;
|
||||
|
||||
import java.time.Duration;
|
||||
import java.util.Objects;
|
||||
import java.util.regex.Pattern;
|
||||
|
||||
/** Atomic inbox claim request with separate processing and terminal retention windows. */
|
||||
public record InboxClaimRequest(
|
||||
InboxScopeDigest scope,
|
||||
String messageIntentDigest,
|
||||
InboxClaimAttempt claimAttempt,
|
||||
Duration processingLease,
|
||||
Duration terminalRetention) {
|
||||
|
||||
private static final Pattern DIGEST = Pattern.compile("[0-9a-f]{64}");
|
||||
|
||||
public InboxClaimRequest {
|
||||
Objects.requireNonNull(scope, "scope");
|
||||
Objects.requireNonNull(messageIntentDigest, "messageIntentDigest");
|
||||
Objects.requireNonNull(claimAttempt, "claimAttempt");
|
||||
if (!DIGEST.matcher(messageIntentDigest).matches()) {
|
||||
throw new IllegalArgumentException(
|
||||
"message intent digest must be a 64-character lowercase hexadecimal value");
|
||||
}
|
||||
requirePositiveBounded("processing lease", processingLease, Duration.ofHours(1));
|
||||
requirePositiveBounded("terminal retention", terminalRetention, Duration.ofDays(30));
|
||||
}
|
||||
|
||||
private static void requirePositiveBounded(String name, Duration value, Duration maximum) {
|
||||
Objects.requireNonNull(value, name);
|
||||
if (value.isZero() || value.isNegative() || value.compareTo(maximum) > 0) {
|
||||
throw new IllegalArgumentException(name + " must be positive and at most " + maximum);
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,32 @@
|
||||
package dev.caskeleton.application.inbox;
|
||||
|
||||
import dev.caskeleton.application.transaction.OperationId;
|
||||
import java.util.Objects;
|
||||
|
||||
/** Full owner/attempt/claim-operation/state-revision CAS tuple for one inbox message. */
|
||||
public record InboxOwner(
|
||||
InboxScopeDigest scope,
|
||||
String ownerToken,
|
||||
long attempt,
|
||||
long stateRevision,
|
||||
OperationId claimOperationId) {
|
||||
|
||||
public InboxOwner {
|
||||
Objects.requireNonNull(scope, "scope");
|
||||
Objects.requireNonNull(ownerToken, "ownerToken");
|
||||
Objects.requireNonNull(claimOperationId, "claimOperationId");
|
||||
if (ownerToken.isBlank() || ownerToken.length() > 128) {
|
||||
throw new IllegalArgumentException("owner token must contain 1-128 characters");
|
||||
}
|
||||
if (attempt < 1) {
|
||||
throw new IllegalArgumentException("attempt must be positive");
|
||||
}
|
||||
if (stateRevision < 0) {
|
||||
throw new IllegalArgumentException("state revision must be non-negative");
|
||||
}
|
||||
}
|
||||
|
||||
public InboxOwner withStateRevision(long revision) {
|
||||
return new InboxOwner(scope, ownerToken, attempt, revision, claimOperationId);
|
||||
}
|
||||
}
|
||||
+20
@@ -0,0 +1,20 @@
|
||||
package dev.caskeleton.application.inbox;
|
||||
|
||||
import java.util.Objects;
|
||||
import java.util.Optional;
|
||||
|
||||
/** Transition outcome and updated owner handle while ownership remains live. */
|
||||
public record InboxOwnerTransition(InboxTransitionOutcome outcome, Optional<InboxOwner> owner) {
|
||||
|
||||
public InboxOwnerTransition {
|
||||
Objects.requireNonNull(outcome, "outcome");
|
||||
Objects.requireNonNull(owner, "owner");
|
||||
boolean mustCarryOwner = outcome == InboxTransitionOutcome.PROCESSING_STARTED;
|
||||
if (mustCarryOwner != owner.isPresent()) {
|
||||
throw new IllegalArgumentException(
|
||||
mustCarryOwner
|
||||
? "PROCESSING_STARTED must carry an updated owner"
|
||||
: outcome + " must not carry an owner");
|
||||
}
|
||||
}
|
||||
}
|
||||
+18
@@ -0,0 +1,18 @@
|
||||
package dev.caskeleton.application.inbox;
|
||||
|
||||
import java.util.Objects;
|
||||
import java.util.regex.Pattern;
|
||||
|
||||
/** Versioned canonical digest of consumer-group, handler, tenant, and message ID scope. */
|
||||
public record InboxScopeDigest(String value) {
|
||||
|
||||
private static final Pattern LOWERCASE_SHA_256 = Pattern.compile("[0-9a-f]{64}");
|
||||
|
||||
public InboxScopeDigest {
|
||||
Objects.requireNonNull(value, "value");
|
||||
if (!LOWERCASE_SHA_256.matcher(value).matches()) {
|
||||
throw new IllegalArgumentException(
|
||||
"inbox scope must be a 64-character lowercase hexadecimal digest");
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,10 @@
|
||||
package dev.caskeleton.application.inbox;
|
||||
|
||||
/** Same-store inbox lifecycle. */
|
||||
public enum InboxState {
|
||||
RECEIVED,
|
||||
PROCESSING,
|
||||
COMPLETED,
|
||||
RETRYABLE,
|
||||
DEAD
|
||||
}
|
||||
+23
@@ -0,0 +1,23 @@
|
||||
package dev.caskeleton.application.inbox;
|
||||
|
||||
import dev.caskeleton.application.transaction.OperationId;
|
||||
import java.time.Duration;
|
||||
|
||||
/**
|
||||
* Owner-safe same-store inbox state machine. Broker ACK must happen only after transaction commit.
|
||||
*/
|
||||
public interface InboxStorePort {
|
||||
|
||||
InboxClaimAttempt newClaimAttempt(OperationId operationId);
|
||||
|
||||
InboxClaimOutcome claim(InboxClaimRequest request);
|
||||
|
||||
InboxOwnerTransition markProcessing(InboxOwner owner, OperationId operationId);
|
||||
|
||||
InboxTransitionOutcome complete(InboxOwner owner, OperationId operationId);
|
||||
|
||||
InboxTransitionOutcome markRetryable(
|
||||
InboxOwner owner, Duration retention, OperationId operationId);
|
||||
|
||||
InboxTransitionOutcome markDead(InboxOwner owner, Duration retention, OperationId operationId);
|
||||
}
|
||||
+15
@@ -0,0 +1,15 @@
|
||||
package dev.caskeleton.application.inbox;
|
||||
|
||||
/** Owner-safe inbox transition classification. */
|
||||
public enum InboxTransitionOutcome {
|
||||
PROCESSING_STARTED,
|
||||
COMPLETED,
|
||||
RETRYABLE,
|
||||
DEAD,
|
||||
ALREADY_APPLIED_SAME_OPERATION,
|
||||
RESULT_CONFLICT,
|
||||
ABSENT,
|
||||
NOT_OWNER,
|
||||
INVALID_STATE,
|
||||
STALE_REVISION
|
||||
}
|
||||
+6
-1
@@ -27,7 +27,12 @@ public record CallBudget(long monotonicDeadlineNanos) {
|
||||
throw new IllegalArgumentException(
|
||||
"call budget duration exceeds the supported range", exception);
|
||||
}
|
||||
return new CallBudget(monotonicNowNanos + durationNanos);
|
||||
try {
|
||||
return new CallBudget(Math.addExact(monotonicNowNanos, durationNanos));
|
||||
} catch (ArithmeticException exception) {
|
||||
throw new IllegalArgumentException(
|
||||
"call budget deadline exceeds the monotonic range", exception);
|
||||
}
|
||||
}
|
||||
|
||||
public long remainingNanosAt(long monotonicNowNanos) {
|
||||
|
||||
+35
@@ -0,0 +1,35 @@
|
||||
package dev.caskeleton.application.outbox.v2;
|
||||
|
||||
import java.time.Instant;
|
||||
import java.util.Objects;
|
||||
|
||||
/** Immutable message envelope plus the mutable delivery owner handle. */
|
||||
public record ClaimedOutboxDelivery(
|
||||
OutboxDeliveryOwner owner,
|
||||
String eventType,
|
||||
int eventSchema,
|
||||
String aggregateType,
|
||||
String aggregateId,
|
||||
long aggregateVersion,
|
||||
int eventOrdinal,
|
||||
String partitionKey,
|
||||
String contentType,
|
||||
String correlationId,
|
||||
String causationId,
|
||||
Instant occurredAt,
|
||||
String payload,
|
||||
String payloadDigest) {
|
||||
|
||||
public ClaimedOutboxDelivery {
|
||||
Objects.requireNonNull(owner, "owner");
|
||||
Objects.requireNonNull(eventType, "eventType");
|
||||
Objects.requireNonNull(aggregateType, "aggregateType");
|
||||
Objects.requireNonNull(aggregateId, "aggregateId");
|
||||
Objects.requireNonNull(partitionKey, "partitionKey");
|
||||
Objects.requireNonNull(contentType, "contentType");
|
||||
Objects.requireNonNull(correlationId, "correlationId");
|
||||
Objects.requireNonNull(occurredAt, "occurredAt");
|
||||
Objects.requireNonNull(payload, "payload");
|
||||
Objects.requireNonNull(payloadDigest, "payloadDigest");
|
||||
}
|
||||
}
|
||||
+66
@@ -0,0 +1,66 @@
|
||||
package dev.caskeleton.application.outbox.v2;
|
||||
|
||||
import java.nio.charset.StandardCharsets;
|
||||
import java.time.Instant;
|
||||
import java.util.Objects;
|
||||
|
||||
/**
|
||||
* Immutable outbox event intent pinned before any whole-transaction retry begins.
|
||||
*
|
||||
* <p>The aggregate version plus deterministic event ordinal is the ordering authority. The stable
|
||||
* partition key is mandatory because this baseline advertises ordered destinations only.
|
||||
*/
|
||||
public record NewOutboxEventV2(
|
||||
String eventId,
|
||||
String aggregateType,
|
||||
String aggregateId,
|
||||
long aggregateVersion,
|
||||
int eventOrdinal,
|
||||
String eventType,
|
||||
int eventSchema,
|
||||
String logicalDestination,
|
||||
String partitionKey,
|
||||
String contentType,
|
||||
String correlationId,
|
||||
String causationId,
|
||||
Instant occurredAt,
|
||||
String payload) {
|
||||
|
||||
private static final int MAXIMUM_PAYLOAD_BYTES = 1024 * 1024;
|
||||
|
||||
public NewOutboxEventV2 {
|
||||
requireBounded("event ID", eventId, 64);
|
||||
requireBounded("aggregate type", aggregateType, 128);
|
||||
requireBounded("aggregate ID", aggregateId, 256);
|
||||
if (aggregateVersion < 1) {
|
||||
throw new IllegalArgumentException("aggregate version must be positive");
|
||||
}
|
||||
if (eventOrdinal < 0 || eventOrdinal > 1023) {
|
||||
throw new IllegalArgumentException("event ordinal must be between 0 and 1023");
|
||||
}
|
||||
requireBounded("event type", eventType, 256);
|
||||
if (eventSchema < 1) {
|
||||
throw new IllegalArgumentException("event schema must be positive");
|
||||
}
|
||||
requireBounded("logical destination", logicalDestination, 256);
|
||||
requireBounded("partition key", partitionKey, 256);
|
||||
requireBounded("content type", contentType, 128);
|
||||
requireBounded("correlation ID", correlationId, 128);
|
||||
if (causationId != null) {
|
||||
requireBounded("causation ID", causationId, 128);
|
||||
}
|
||||
Objects.requireNonNull(occurredAt, "occurredAt");
|
||||
Objects.requireNonNull(payload, "payload");
|
||||
if (payload.isBlank()
|
||||
|| payload.getBytes(StandardCharsets.UTF_8).length > MAXIMUM_PAYLOAD_BYTES) {
|
||||
throw new IllegalArgumentException(
|
||||
"payload must be non-blank and at most " + MAXIMUM_PAYLOAD_BYTES + " UTF-8 bytes");
|
||||
}
|
||||
}
|
||||
|
||||
private static void requireBounded(String name, String value, int maximumLength) {
|
||||
if (value == null || value.isBlank() || value.length() > maximumLength) {
|
||||
throw new IllegalArgumentException(name + " must contain 1-" + maximumLength + " characters");
|
||||
}
|
||||
}
|
||||
}
|
||||
+9
@@ -0,0 +1,9 @@
|
||||
package dev.caskeleton.application.outbox.v2;
|
||||
|
||||
/** Conflict-safe result of immutable identity plus event-envelope append. */
|
||||
public enum OutboxAppendOutcome {
|
||||
APPENDED,
|
||||
ALREADY_APPENDED_SAME_EVENT,
|
||||
EVENT_ID_CONFLICT,
|
||||
AGGREGATE_ORDER_CONFLICT
|
||||
}
|
||||
+12
@@ -0,0 +1,12 @@
|
||||
package dev.caskeleton.application.outbox.v2;
|
||||
|
||||
/**
|
||||
* Same-store immutable outbox append.
|
||||
*
|
||||
* <p>The implementation must participate in the caller's active primary read-write transaction; it
|
||||
* must never open a repository-local transaction.
|
||||
*/
|
||||
public interface OutboxAppendPortV2 {
|
||||
|
||||
OutboxAppendReceipt append(NewOutboxEventV2 event);
|
||||
}
|
||||
+23
@@ -0,0 +1,23 @@
|
||||
package dev.caskeleton.application.outbox.v2;
|
||||
|
||||
import java.time.LocalDate;
|
||||
import java.util.Objects;
|
||||
|
||||
/** Database-authoritative routing receipt for one immutable event append. */
|
||||
public record OutboxAppendReceipt(
|
||||
OutboxAppendOutcome outcome,
|
||||
String eventId,
|
||||
LocalDate retentionBucket,
|
||||
long publicationEpoch,
|
||||
OutboxDispatchAuthority dispatchAuthority) {
|
||||
|
||||
public OutboxAppendReceipt {
|
||||
Objects.requireNonNull(outcome, "outcome");
|
||||
Objects.requireNonNull(eventId, "eventId");
|
||||
Objects.requireNonNull(retentionBucket, "retentionBucket");
|
||||
Objects.requireNonNull(dispatchAuthority, "dispatchAuthority");
|
||||
if (publicationEpoch < 1) {
|
||||
throw new IllegalArgumentException("publication epoch must be positive");
|
||||
}
|
||||
}
|
||||
}
|
||||
+26
@@ -0,0 +1,26 @@
|
||||
package dev.caskeleton.application.outbox.v2;
|
||||
|
||||
import java.time.Duration;
|
||||
|
||||
/** Bounded strict-order polling claim request. */
|
||||
public record OutboxDeliveryClaimRequest(
|
||||
String destination, String claimOwner, int batchSize, Duration claimLease) {
|
||||
|
||||
public OutboxDeliveryClaimRequest {
|
||||
if (destination == null || destination.isBlank()) {
|
||||
throw new IllegalArgumentException("destination must be present");
|
||||
}
|
||||
if (claimOwner == null || claimOwner.isBlank() || claimOwner.length() > 128) {
|
||||
throw new IllegalArgumentException("claim owner must contain 1-128 characters");
|
||||
}
|
||||
if (batchSize < 1 || batchSize > 100) {
|
||||
throw new IllegalArgumentException("batch size must be between 1 and 100");
|
||||
}
|
||||
if (claimLease == null
|
||||
|| claimLease.isZero()
|
||||
|| claimLease.isNegative()
|
||||
|| claimLease.compareTo(Duration.ofMinutes(5)) > 0) {
|
||||
throw new IllegalArgumentException("claim lease must be positive and at most PT5M");
|
||||
}
|
||||
}
|
||||
}
|
||||
+45
@@ -0,0 +1,45 @@
|
||||
package dev.caskeleton.application.outbox.v2;
|
||||
|
||||
import java.time.LocalDate;
|
||||
import java.util.Objects;
|
||||
import java.util.regex.Pattern;
|
||||
|
||||
/** Full owner/token/status-version CAS handle for one destination delivery. */
|
||||
public record OutboxDeliveryOwner(
|
||||
LocalDate retentionBucket,
|
||||
String eventId,
|
||||
String destination,
|
||||
String claimOwner,
|
||||
String claimToken,
|
||||
int attempt,
|
||||
long version,
|
||||
long publicationEpoch) {
|
||||
|
||||
private static final Pattern CLAIM_TOKEN = Pattern.compile("[0-9a-f]{64}");
|
||||
|
||||
public OutboxDeliveryOwner {
|
||||
Objects.requireNonNull(retentionBucket, "retentionBucket");
|
||||
requirePresent(eventId, "event ID");
|
||||
requirePresent(destination, "destination");
|
||||
requirePresent(claimOwner, "claim owner");
|
||||
if (claimToken == null || !CLAIM_TOKEN.matcher(claimToken).matches()) {
|
||||
throw new IllegalArgumentException(
|
||||
"claim token must be a 64-character lowercase hexadecimal value");
|
||||
}
|
||||
if (attempt < 1) {
|
||||
throw new IllegalArgumentException("attempt must be positive");
|
||||
}
|
||||
if (version < 1) {
|
||||
throw new IllegalArgumentException("version must be positive");
|
||||
}
|
||||
if (publicationEpoch < 1) {
|
||||
throw new IllegalArgumentException("publication epoch must be positive");
|
||||
}
|
||||
}
|
||||
|
||||
private static void requirePresent(String value, String name) {
|
||||
if (value == null || value.isBlank()) {
|
||||
throw new IllegalArgumentException(name + " must be present");
|
||||
}
|
||||
}
|
||||
}
|
||||
+13
@@ -0,0 +1,13 @@
|
||||
package dev.caskeleton.application.outbox.v2;
|
||||
|
||||
import dev.caskeleton.application.transaction.OperationId;
|
||||
import java.util.Objects;
|
||||
|
||||
/** Idempotent transition request for a claimed delivery. */
|
||||
public record OutboxDeliveryTransition(OutboxDeliveryOwner owner, OperationId operationId) {
|
||||
|
||||
public OutboxDeliveryTransition {
|
||||
Objects.requireNonNull(owner, "owner");
|
||||
Objects.requireNonNull(operationId, "operationId");
|
||||
}
|
||||
}
|
||||
+15
@@ -0,0 +1,15 @@
|
||||
package dev.caskeleton.application.outbox.v2;
|
||||
|
||||
/** Owner-safe polling completion/failure result. */
|
||||
public enum OutboxDeliveryTransitionOutcome {
|
||||
PUBLISHED,
|
||||
RETRY_SCHEDULED,
|
||||
DEAD,
|
||||
ALREADY_APPLIED_SAME_OPERATION,
|
||||
RESULT_CONFLICT,
|
||||
ABSENT,
|
||||
NOT_OWNER,
|
||||
NOT_CLAIMED,
|
||||
STALE_VERSION,
|
||||
AUTHORITY_MISMATCH
|
||||
}
|
||||
+8
@@ -0,0 +1,8 @@
|
||||
package dev.caskeleton.application.outbox.v2;
|
||||
|
||||
/** Per-row dispatch authority derived from the locked publication control row. */
|
||||
public enum OutboxDispatchAuthority {
|
||||
LEGACY_SHADOW,
|
||||
POLLING_V2,
|
||||
CDC
|
||||
}
|
||||
+17
@@ -0,0 +1,17 @@
|
||||
package dev.caskeleton.application.outbox.v2;
|
||||
|
||||
import java.time.Instant;
|
||||
import java.util.List;
|
||||
|
||||
/** Owner-safe polling delivery state machine; broker publish occurs outside its transactions. */
|
||||
public interface OutboxPollingDeliveryPortV2 {
|
||||
|
||||
List<ClaimedOutboxDelivery> claimBatch(OutboxDeliveryClaimRequest request);
|
||||
|
||||
OutboxDeliveryTransitionOutcome markPublished(OutboxDeliveryTransition transition);
|
||||
|
||||
OutboxDeliveryTransitionOutcome markRetryable(
|
||||
OutboxDeliveryTransition transition, Instant nextAttemptAt, String errorCode);
|
||||
|
||||
OutboxDeliveryTransitionOutcome markDead(OutboxDeliveryTransition transition, String errorCode);
|
||||
}
|
||||
+8
@@ -0,0 +1,8 @@
|
||||
package dev.caskeleton.application.outbox.v2;
|
||||
|
||||
/** The single database-controlled publisher authority for the primary outbox scope. */
|
||||
public enum OutboxPublicationAuthority {
|
||||
LEGACY_POLLING,
|
||||
POLLING_V2,
|
||||
CDC
|
||||
}
|
||||
+26
@@ -0,0 +1,26 @@
|
||||
package dev.caskeleton.application.transaction;
|
||||
|
||||
import java.util.Objects;
|
||||
|
||||
/**
|
||||
* Stable, caller-owned identity for one logical write operation.
|
||||
*
|
||||
* <p>The value is intentionally opaque. Adapters may use it for reconciliation, but must not invent
|
||||
* a replacement identity after an uncertain commit.
|
||||
*/
|
||||
public record OperationId(String value) {
|
||||
|
||||
private static final int MAXIMUM_LENGTH = 128;
|
||||
|
||||
public OperationId {
|
||||
Objects.requireNonNull(value, "value must be non-null");
|
||||
if (value.isBlank() || value.length() > MAXIMUM_LENGTH || !isPrintableAscii(value)) {
|
||||
throw new IllegalArgumentException(
|
||||
"operation ID must contain 1-128 printable non-whitespace ASCII characters");
|
||||
}
|
||||
}
|
||||
|
||||
private static boolean isPrintableAscii(String value) {
|
||||
return value.chars().allMatch(character -> character >= 0x21 && character <= 0x7e);
|
||||
}
|
||||
}
|
||||
+13
@@ -0,0 +1,13 @@
|
||||
package dev.caskeleton.application.transaction;
|
||||
|
||||
import java.util.function.Supplier;
|
||||
|
||||
/**
|
||||
* Additive transaction port for named policies and explicit outcomes.
|
||||
*
|
||||
* <p>{@link TransactionPort} remains source-compatible for existing callers and fakes.
|
||||
*/
|
||||
public interface PolicyTransactionPort extends TransactionPort {
|
||||
|
||||
<T> TransactionResult<T> inTransaction(TransactionRequest request, Supplier<T> action);
|
||||
}
|
||||
+9
@@ -0,0 +1,9 @@
|
||||
package dev.caskeleton.application.transaction;
|
||||
|
||||
/** Application-owned read consistency vocabulary. */
|
||||
public enum ReadConsistency {
|
||||
STRONG,
|
||||
READ_YOUR_WRITES,
|
||||
EVENTUAL,
|
||||
BOUNDED_STALENESS
|
||||
}
|
||||
+24
@@ -0,0 +1,24 @@
|
||||
package dev.caskeleton.application.transaction;
|
||||
|
||||
import java.util.Objects;
|
||||
|
||||
/**
|
||||
* Sanitized, bounded reference that an operator or application workflow can use to reconcile an
|
||||
* uncertain transaction outcome.
|
||||
*/
|
||||
public record ReconciliationReference(String value) {
|
||||
|
||||
private static final int MAXIMUM_LENGTH = 256;
|
||||
|
||||
public ReconciliationReference {
|
||||
Objects.requireNonNull(value, "value must be non-null");
|
||||
if (value.isBlank() || value.length() > MAXIMUM_LENGTH || !isPrintableAscii(value)) {
|
||||
throw new IllegalArgumentException(
|
||||
"reconciliation reference must contain 1-256 printable non-whitespace ASCII characters");
|
||||
}
|
||||
}
|
||||
|
||||
private static boolean isPrintableAscii(String value) {
|
||||
return value.chars().allMatch(character -> character >= 0x21 && character <= 0x7e);
|
||||
}
|
||||
}
|
||||
+16
@@ -0,0 +1,16 @@
|
||||
package dev.caskeleton.application.transaction;
|
||||
|
||||
/**
|
||||
* A transaction was rejected before application work started because its policy, route, or
|
||||
* remaining deadline could not be honored.
|
||||
*/
|
||||
public final class TransactionAdmissionException extends RuntimeException {
|
||||
|
||||
public TransactionAdmissionException(String message) {
|
||||
super(message);
|
||||
}
|
||||
|
||||
public TransactionAdmissionException(String message, Throwable cause) {
|
||||
super(message, cause);
|
||||
}
|
||||
}
|
||||
+10
@@ -0,0 +1,10 @@
|
||||
package dev.caskeleton.application.transaction;
|
||||
|
||||
/** Framework-neutral outcome of an application transaction boundary. */
|
||||
public enum TransactionOutcome {
|
||||
COMMITTED,
|
||||
PARTICIPATING_PENDING_OUTER,
|
||||
DETERMINATE_ROLLBACK,
|
||||
INDETERMINATE,
|
||||
COMMITTED_WITH_POST_COMMIT_FAILURE
|
||||
}
|
||||
+12
@@ -0,0 +1,12 @@
|
||||
package dev.caskeleton.application.transaction;
|
||||
|
||||
/** Last safely observed phase of a physical transaction. */
|
||||
public enum TransactionPhase {
|
||||
ROUTE_ADMISSION,
|
||||
CONNECTION_ACQUIRED,
|
||||
ACTIVE,
|
||||
FLUSHED,
|
||||
COMMIT_REQUESTED,
|
||||
COMMIT_ACKED,
|
||||
SYNCHRONIZATION_CLEANUP
|
||||
}
|
||||
+45
@@ -0,0 +1,45 @@
|
||||
package dev.caskeleton.application.transaction;
|
||||
|
||||
/**
|
||||
* Allowlisted application transaction policies.
|
||||
*
|
||||
* <p>Callers select semantic policy IDs rather than framework propagation, isolation, route, or
|
||||
* timeout numbers. Legacy facade policies remain adapter-internal and are intentionally absent.
|
||||
*/
|
||||
public enum TransactionPolicyId {
|
||||
COMMAND_DEFAULT(false, true),
|
||||
COMMAND_SERIALIZABLE_REPLAY_SAFE(false, true),
|
||||
QUERY_PRIMARY(true, false),
|
||||
QUERY_REPLICA_ELIGIBLE(true, false),
|
||||
OUTBOX_APPEND(false, false),
|
||||
INBOX_AND_HANDLER(false, true),
|
||||
MAINTENANCE_NEW(false, true);
|
||||
|
||||
private final boolean readPolicy;
|
||||
private final boolean operationIdRequired;
|
||||
|
||||
TransactionPolicyId(boolean readPolicy, boolean operationIdRequired) {
|
||||
this.readPolicy = readPolicy;
|
||||
this.operationIdRequired = operationIdRequired;
|
||||
}
|
||||
|
||||
public boolean isReadPolicy() {
|
||||
return readPolicy;
|
||||
}
|
||||
|
||||
public boolean requiresOperationId() {
|
||||
return operationIdRequired;
|
||||
}
|
||||
|
||||
public boolean supports(ReadConsistency consistency) {
|
||||
if (this == QUERY_PRIMARY) {
|
||||
return consistency == ReadConsistency.STRONG
|
||||
|| consistency == ReadConsistency.READ_YOUR_WRITES;
|
||||
}
|
||||
if (this == QUERY_REPLICA_ELIGIBLE) {
|
||||
return consistency == ReadConsistency.EVENTUAL
|
||||
|| consistency == ReadConsistency.BOUNDED_STALENESS;
|
||||
}
|
||||
return false;
|
||||
}
|
||||
}
|
||||
+38
@@ -0,0 +1,38 @@
|
||||
package dev.caskeleton.application.transaction;
|
||||
|
||||
import dev.caskeleton.application.outbound.CallBudget;
|
||||
import java.util.Objects;
|
||||
import java.util.Optional;
|
||||
|
||||
/** One framework-neutral request to execute an allowlisted transaction policy. */
|
||||
public record TransactionRequest(
|
||||
TransactionPolicyId policyId,
|
||||
CallBudget callBudget,
|
||||
Optional<ReadConsistency> readConsistency,
|
||||
Optional<OperationId> operationId) {
|
||||
|
||||
public TransactionRequest {
|
||||
Objects.requireNonNull(policyId, "policyId must be non-null");
|
||||
Objects.requireNonNull(callBudget, "callBudget must be non-null");
|
||||
Objects.requireNonNull(readConsistency, "readConsistency must be non-null");
|
||||
Objects.requireNonNull(operationId, "operationId must be non-null");
|
||||
|
||||
if (policyId.isReadPolicy()) {
|
||||
ReadConsistency selected =
|
||||
readConsistency.orElseThrow(
|
||||
() ->
|
||||
new IllegalArgumentException(
|
||||
"readConsistency is required for " + policyId.name()));
|
||||
if (!policyId.supports(selected)) {
|
||||
throw new IllegalArgumentException(
|
||||
"readConsistency " + selected + " is not allowed for " + policyId.name());
|
||||
}
|
||||
} else if (readConsistency.isPresent()) {
|
||||
throw new IllegalArgumentException("readConsistency is forbidden for " + policyId.name());
|
||||
}
|
||||
|
||||
if (policyId.requiresOperationId() && operationId.isEmpty()) {
|
||||
throw new IllegalArgumentException("operationId is required for " + policyId.name());
|
||||
}
|
||||
}
|
||||
}
|
||||
+83
@@ -0,0 +1,83 @@
|
||||
package dev.caskeleton.application.transaction;
|
||||
|
||||
import java.util.Objects;
|
||||
import java.util.Optional;
|
||||
|
||||
/**
|
||||
* Outcome algebra for policy-based transactions.
|
||||
*
|
||||
* <p>A participant result never claims commit. An indeterminate result never grants replay
|
||||
* authority.
|
||||
*/
|
||||
public sealed interface TransactionResult<T> {
|
||||
|
||||
TransactionOutcome outcome();
|
||||
|
||||
record Committed<T>(T value, Optional<OperationId> operationId) implements TransactionResult<T> {
|
||||
|
||||
public Committed {
|
||||
Objects.requireNonNull(operationId, "operationId must be non-null");
|
||||
}
|
||||
|
||||
@Override
|
||||
public TransactionOutcome outcome() {
|
||||
return TransactionOutcome.COMMITTED;
|
||||
}
|
||||
}
|
||||
|
||||
record Participating<T>(T value) implements TransactionResult<T> {
|
||||
|
||||
@Override
|
||||
public TransactionOutcome outcome() {
|
||||
return TransactionOutcome.PARTICIPATING_PENDING_OUTER;
|
||||
}
|
||||
}
|
||||
|
||||
record DeterminateRollback<T>(RuntimeException failure) implements TransactionResult<T> {
|
||||
|
||||
public DeterminateRollback {
|
||||
Objects.requireNonNull(failure, "failure must be non-null");
|
||||
}
|
||||
|
||||
@Override
|
||||
public TransactionOutcome outcome() {
|
||||
return TransactionOutcome.DETERMINATE_ROLLBACK;
|
||||
}
|
||||
}
|
||||
|
||||
record Indeterminate<T>(
|
||||
Optional<OperationId> operationId,
|
||||
TransactionPhase lastObservedPhase,
|
||||
Optional<ReconciliationReference> reconciliationReference)
|
||||
implements TransactionResult<T> {
|
||||
|
||||
public Indeterminate {
|
||||
Objects.requireNonNull(operationId, "operationId must be non-null");
|
||||
Objects.requireNonNull(lastObservedPhase, "lastObservedPhase must be non-null");
|
||||
Objects.requireNonNull(reconciliationReference, "reconciliationReference must be non-null");
|
||||
if (operationId.isEmpty() && reconciliationReference.isPresent()) {
|
||||
throw new IllegalArgumentException("reconciliationReference requires a stable operationId");
|
||||
}
|
||||
}
|
||||
|
||||
@Override
|
||||
public TransactionOutcome outcome() {
|
||||
return TransactionOutcome.INDETERMINATE;
|
||||
}
|
||||
}
|
||||
|
||||
record CommittedWithPostCommitFailure<T>(
|
||||
T value, Optional<OperationId> operationId, RuntimeException operationalFailure)
|
||||
implements TransactionResult<T> {
|
||||
|
||||
public CommittedWithPostCommitFailure {
|
||||
Objects.requireNonNull(operationId, "operationId must be non-null");
|
||||
Objects.requireNonNull(operationalFailure, "operationalFailure must be non-null");
|
||||
}
|
||||
|
||||
@Override
|
||||
public TransactionOutcome outcome() {
|
||||
return TransactionOutcome.COMMITTED_WITH_POST_COMMIT_FAILURE;
|
||||
}
|
||||
}
|
||||
}
|
||||
+91
@@ -0,0 +1,91 @@
|
||||
package dev.caskeleton.application.idempotency.v2;
|
||||
|
||||
import static org.assertj.core.api.Assertions.assertThat;
|
||||
import static org.assertj.core.api.Assertions.assertThatThrownBy;
|
||||
|
||||
import dev.caskeleton.application.idempotency.RequestFingerprint;
|
||||
import dev.caskeleton.application.transaction.OperationId;
|
||||
import java.nio.charset.StandardCharsets;
|
||||
import java.time.Duration;
|
||||
import org.junit.jupiter.api.Test;
|
||||
|
||||
class IdempotencyV2ContractTest {
|
||||
|
||||
private static final IdempotencyScopeDigest SCOPE =
|
||||
new IdempotencyScopeDigest("a".repeat(64), 3, "CREATE_WORK_LOG");
|
||||
private static final RequestFingerprint FINGERPRINT =
|
||||
RequestFingerprint.ofSha256("request".getBytes(StandardCharsets.UTF_8));
|
||||
private static final IdempotencyClaimAttempt ATTEMPT =
|
||||
new IdempotencyClaimAttempt("b".repeat(64), new OperationId("claim-1"));
|
||||
|
||||
@Test
|
||||
void scopeAcceptsOnlyCanonicalLowercaseSha256AndPositiveKeyVersion() {
|
||||
assertThat(SCOPE.digest()).hasSize(64);
|
||||
|
||||
assertThatThrownBy(() -> new IdempotencyScopeDigest("A".repeat(64), 1, "CREATE_WORK_LOG"))
|
||||
.isInstanceOf(IllegalArgumentException.class)
|
||||
.hasMessageContaining("lowercase");
|
||||
assertThatThrownBy(() -> new IdempotencyScopeDigest("a".repeat(64), 0, "CREATE_WORK_LOG"))
|
||||
.isInstanceOf(IllegalArgumentException.class)
|
||||
.hasMessageContaining("key digest version");
|
||||
}
|
||||
|
||||
@Test
|
||||
void claimRequestSeparatesFiniteProcessingAndReplayTtls() {
|
||||
IdempotencyClaimRequest request =
|
||||
new IdempotencyClaimRequest(
|
||||
SCOPE,
|
||||
FINGERPRINT,
|
||||
ATTEMPT,
|
||||
Duration.ofSeconds(30),
|
||||
Duration.ofHours(24),
|
||||
"json.v1",
|
||||
2);
|
||||
|
||||
assertThat(request.processingLeaseTtl()).isEqualTo(Duration.ofSeconds(30));
|
||||
assertThat(request.replayTtl()).isEqualTo(Duration.ofHours(24));
|
||||
|
||||
assertThatThrownBy(
|
||||
() ->
|
||||
new IdempotencyClaimRequest(
|
||||
SCOPE, FINGERPRINT, ATTEMPT, Duration.ZERO, Duration.ofHours(24), "json.v1", 2))
|
||||
.isInstanceOf(IllegalArgumentException.class)
|
||||
.hasMessageContaining("processing lease TTL");
|
||||
}
|
||||
|
||||
@Test
|
||||
void ownerCarriesTheFullCasTuple() {
|
||||
IdempotencyOwner owner =
|
||||
new IdempotencyOwner(SCOPE, ATTEMPT.ownerToken(), 2, 7, ATTEMPT.operationId());
|
||||
|
||||
assertThat(owner.attempt()).isEqualTo(2);
|
||||
assertThat(owner.stateRevision()).isEqualTo(7);
|
||||
assertThat(owner.claimOperationId()).isEqualTo(new OperationId("claim-1"));
|
||||
|
||||
assertThatThrownBy(
|
||||
() -> new IdempotencyOwner(SCOPE, ATTEMPT.ownerToken(), 0, 7, ATTEMPT.operationId()))
|
||||
.isInstanceOf(IllegalArgumentException.class)
|
||||
.hasMessageContaining("attempt");
|
||||
}
|
||||
|
||||
@Test
|
||||
void transitionResultRequiresAnOwnerOnlyForSuccessfulOwnerTransitions() {
|
||||
IdempotencyOwner owner =
|
||||
new IdempotencyOwner(SCOPE, ATTEMPT.ownerToken(), 1, 1, ATTEMPT.operationId());
|
||||
|
||||
assertThat(
|
||||
new IdempotencyMutationResult<>(
|
||||
IdempotencyStartOutcome.STARTED, owner, IdempotencyStartOutcome::carriesOwner)
|
||||
.owner())
|
||||
.contains(owner);
|
||||
|
||||
assertThatThrownBy(
|
||||
() ->
|
||||
new IdempotencyMutationResult<>(
|
||||
IdempotencyStartOutcome.NOT_OWNER,
|
||||
owner,
|
||||
IdempotencyStartOutcome::carriesOwner))
|
||||
.isInstanceOf(IllegalArgumentException.class)
|
||||
.hasMessageContaining("must not carry");
|
||||
}
|
||||
}
|
||||
+39
@@ -0,0 +1,39 @@
|
||||
package dev.caskeleton.application.inbox;
|
||||
|
||||
import static org.assertj.core.api.Assertions.assertThat;
|
||||
import static org.assertj.core.api.Assertions.assertThatThrownBy;
|
||||
|
||||
import dev.caskeleton.application.transaction.OperationId;
|
||||
import java.time.Duration;
|
||||
import org.junit.jupiter.api.Test;
|
||||
|
||||
class InboxContractTest {
|
||||
|
||||
@Test
|
||||
void scopeIsAnOpaqueCanonicalDigestAndOwnerCarriesTheFullCasTuple() {
|
||||
InboxScopeDigest scope = new InboxScopeDigest("a".repeat(64));
|
||||
InboxOwner owner =
|
||||
new InboxOwner(scope, "b".repeat(64), 2, 7, new OperationId("claim-message-1"));
|
||||
|
||||
assertThat(owner.attempt()).isEqualTo(2);
|
||||
assertThat(owner.stateRevision()).isEqualTo(7);
|
||||
|
||||
assertThatThrownBy(() -> new InboxScopeDigest("A".repeat(64)))
|
||||
.isInstanceOf(IllegalArgumentException.class)
|
||||
.hasMessageContaining("lowercase");
|
||||
}
|
||||
|
||||
@Test
|
||||
void claimRequestSeparatesProcessingLeaseFromTerminalRetention() {
|
||||
InboxClaimRequest request =
|
||||
new InboxClaimRequest(
|
||||
new InboxScopeDigest("a".repeat(64)),
|
||||
"c".repeat(64),
|
||||
new InboxClaimAttempt("b".repeat(64), new OperationId("claim-message-1")),
|
||||
Duration.ofSeconds(30),
|
||||
Duration.ofDays(7));
|
||||
|
||||
assertThat(request.processingLease()).isEqualTo(Duration.ofSeconds(30));
|
||||
assertThat(request.terminalRetention()).isEqualTo(Duration.ofDays(7));
|
||||
}
|
||||
}
|
||||
+7
@@ -33,4 +33,11 @@ class CallBudgetTest {
|
||||
assertThatThrownBy(() -> CallBudget.after(1_000, Duration.ofDays(366)))
|
||||
.isInstanceOf(IllegalArgumentException.class);
|
||||
}
|
||||
|
||||
@Test
|
||||
void rejectsMonotonicDeadlineOverflow() {
|
||||
assertThatThrownBy(() -> CallBudget.after(Long.MAX_VALUE - 10, Duration.ofNanos(11)))
|
||||
.isInstanceOf(IllegalArgumentException.class)
|
||||
.hasMessageContaining("deadline");
|
||||
}
|
||||
}
|
||||
|
||||
+61
@@ -0,0 +1,61 @@
|
||||
package dev.caskeleton.application.outbox.v2;
|
||||
|
||||
import static org.assertj.core.api.Assertions.assertThat;
|
||||
import static org.assertj.core.api.Assertions.assertThatThrownBy;
|
||||
|
||||
import dev.caskeleton.application.transaction.OperationId;
|
||||
import java.time.LocalDate;
|
||||
import org.junit.jupiter.api.Test;
|
||||
|
||||
class OutboxDeliveryV2ContractTest {
|
||||
|
||||
@Test
|
||||
void ownerHandleCarriesTheFullDeliveryCasTuple() {
|
||||
OutboxDeliveryOwner owner =
|
||||
new OutboxDeliveryOwner(
|
||||
LocalDate.parse("2026-07-28"),
|
||||
"event-1",
|
||||
"portfolio.events",
|
||||
"relay-1",
|
||||
"a".repeat(64),
|
||||
2,
|
||||
7,
|
||||
3);
|
||||
|
||||
assertThat(owner.attempt()).isEqualTo(2);
|
||||
assertThat(owner.version()).isEqualTo(7);
|
||||
assertThat(owner.publicationEpoch()).isEqualTo(3);
|
||||
|
||||
assertThatThrownBy(
|
||||
() ->
|
||||
new OutboxDeliveryOwner(
|
||||
LocalDate.parse("2026-07-28"),
|
||||
"event-1",
|
||||
"portfolio.events",
|
||||
"relay-1",
|
||||
"a".repeat(64),
|
||||
0,
|
||||
7,
|
||||
3))
|
||||
.isInstanceOf(IllegalArgumentException.class)
|
||||
.hasMessageContaining("attempt");
|
||||
}
|
||||
|
||||
@Test
|
||||
void completionRequestRequiresStableOperationIdentity() {
|
||||
OutboxDeliveryTransition transition =
|
||||
new OutboxDeliveryTransition(
|
||||
new OutboxDeliveryOwner(
|
||||
LocalDate.parse("2026-07-28"),
|
||||
"event-1",
|
||||
"portfolio.events",
|
||||
"relay-1",
|
||||
"a".repeat(64),
|
||||
1,
|
||||
2,
|
||||
1),
|
||||
new OperationId("publish-complete-1"));
|
||||
|
||||
assertThat(transition.operationId().value()).isEqualTo("publish-complete-1");
|
||||
}
|
||||
}
|
||||
+76
@@ -0,0 +1,76 @@
|
||||
package dev.caskeleton.application.outbox.v2;
|
||||
|
||||
import static org.assertj.core.api.Assertions.assertThat;
|
||||
import static org.assertj.core.api.Assertions.assertThatThrownBy;
|
||||
|
||||
import java.time.Instant;
|
||||
import org.junit.jupiter.api.Test;
|
||||
|
||||
class OutboxV2ContractTest {
|
||||
|
||||
@Test
|
||||
void eventRequiresPinnedAggregateVersionOrdinalAndBoundedEnvelopeIdentity() {
|
||||
NewOutboxEventV2 event =
|
||||
new NewOutboxEventV2(
|
||||
"event-1",
|
||||
"WorkLog",
|
||||
"work-log-42",
|
||||
7,
|
||||
0,
|
||||
"WorkLogCreated",
|
||||
1,
|
||||
"portfolio.events",
|
||||
"work-log-42",
|
||||
"application/json",
|
||||
"correlation-1",
|
||||
null,
|
||||
Instant.parse("2026-07-28T12:00:00Z"),
|
||||
"{\"id\":\"42\"}");
|
||||
|
||||
assertThat(event.aggregateVersion()).isEqualTo(7);
|
||||
assertThat(event.eventOrdinal()).isZero();
|
||||
|
||||
assertThatThrownBy(
|
||||
() ->
|
||||
new NewOutboxEventV2(
|
||||
"event-1",
|
||||
"WorkLog",
|
||||
"work-log-42",
|
||||
0,
|
||||
0,
|
||||
"WorkLogCreated",
|
||||
1,
|
||||
"portfolio.events",
|
||||
"work-log-42",
|
||||
"application/json",
|
||||
"correlation-1",
|
||||
null,
|
||||
Instant.parse("2026-07-28T12:00:00Z"),
|
||||
"{}"))
|
||||
.isInstanceOf(IllegalArgumentException.class)
|
||||
.hasMessageContaining("aggregate version");
|
||||
}
|
||||
|
||||
@Test
|
||||
void orderedDestinationDescriptorRequiresAStablePartitionKey() {
|
||||
assertThatThrownBy(
|
||||
() ->
|
||||
new NewOutboxEventV2(
|
||||
"event-1",
|
||||
"WorkLog",
|
||||
"work-log-42",
|
||||
1,
|
||||
0,
|
||||
"WorkLogCreated",
|
||||
1,
|
||||
"portfolio.events",
|
||||
null,
|
||||
"application/json",
|
||||
"correlation-1",
|
||||
null,
|
||||
Instant.parse("2026-07-28T12:00:00Z"),
|
||||
"{}"))
|
||||
.isInstanceOf(IllegalArgumentException.class)
|
||||
.hasMessageContaining("partition key");
|
||||
}
|
||||
}
|
||||
+25
@@ -0,0 +1,25 @@
|
||||
package dev.caskeleton.application.transaction;
|
||||
|
||||
import static org.assertj.core.api.Assertions.assertThat;
|
||||
import static org.assertj.core.api.Assertions.assertThatThrownBy;
|
||||
|
||||
import org.junit.jupiter.api.Test;
|
||||
|
||||
class OperationIdTest {
|
||||
|
||||
@Test
|
||||
void preservesAnOpaqueStableIdentifier() {
|
||||
OperationId operationId = new OperationId("job-20260728-item-42");
|
||||
|
||||
assertThat(operationId.value()).isEqualTo("job-20260728-item-42");
|
||||
}
|
||||
|
||||
@Test
|
||||
void rejectsBlankControlCharactersAndUnboundedValues() {
|
||||
assertThatThrownBy(() -> new OperationId(" ")).isInstanceOf(IllegalArgumentException.class);
|
||||
assertThatThrownBy(() -> new OperationId("operation\nid"))
|
||||
.isInstanceOf(IllegalArgumentException.class);
|
||||
assertThatThrownBy(() -> new OperationId("a".repeat(129)))
|
||||
.isInstanceOf(IllegalArgumentException.class);
|
||||
}
|
||||
}
|
||||
+114
@@ -0,0 +1,114 @@
|
||||
package dev.caskeleton.application.transaction;
|
||||
|
||||
import static org.assertj.core.api.Assertions.assertThat;
|
||||
import static org.assertj.core.api.Assertions.assertThatThrownBy;
|
||||
|
||||
import dev.caskeleton.application.outbound.CallBudget;
|
||||
import java.time.Duration;
|
||||
import java.util.Optional;
|
||||
import org.junit.jupiter.api.Test;
|
||||
|
||||
class TransactionRequestTest {
|
||||
|
||||
private static final CallBudget BUDGET = CallBudget.after(1_000, Duration.ofSeconds(5));
|
||||
private static final OperationId OPERATION_ID = new OperationId("operation-42");
|
||||
|
||||
@Test
|
||||
void commandPoliciesRequireAStableOperationIdAndRejectReadConsistency() {
|
||||
TransactionRequest request =
|
||||
new TransactionRequest(
|
||||
TransactionPolicyId.COMMAND_DEFAULT,
|
||||
BUDGET,
|
||||
Optional.empty(),
|
||||
Optional.of(OPERATION_ID));
|
||||
|
||||
assertThat(request.operationId()).contains(OPERATION_ID);
|
||||
assertThatThrownBy(
|
||||
() ->
|
||||
new TransactionRequest(
|
||||
TransactionPolicyId.COMMAND_DEFAULT,
|
||||
BUDGET,
|
||||
Optional.empty(),
|
||||
Optional.empty()))
|
||||
.isInstanceOf(IllegalArgumentException.class)
|
||||
.hasMessageContaining("operationId");
|
||||
assertThatThrownBy(
|
||||
() ->
|
||||
new TransactionRequest(
|
||||
TransactionPolicyId.COMMAND_DEFAULT,
|
||||
BUDGET,
|
||||
Optional.of(ReadConsistency.STRONG),
|
||||
Optional.of(OPERATION_ID)))
|
||||
.isInstanceOf(IllegalArgumentException.class)
|
||||
.hasMessageContaining("readConsistency");
|
||||
}
|
||||
|
||||
@Test
|
||||
void queryPrimaryAcceptsOnlyStrongOrReadYourWrites() {
|
||||
assertThat(
|
||||
new TransactionRequest(
|
||||
TransactionPolicyId.QUERY_PRIMARY,
|
||||
BUDGET,
|
||||
Optional.of(ReadConsistency.STRONG),
|
||||
Optional.empty())
|
||||
.readConsistency())
|
||||
.contains(ReadConsistency.STRONG);
|
||||
assertThat(
|
||||
new TransactionRequest(
|
||||
TransactionPolicyId.QUERY_PRIMARY,
|
||||
BUDGET,
|
||||
Optional.of(ReadConsistency.READ_YOUR_WRITES),
|
||||
Optional.empty())
|
||||
.readConsistency())
|
||||
.contains(ReadConsistency.READ_YOUR_WRITES);
|
||||
assertThatThrownBy(
|
||||
() ->
|
||||
new TransactionRequest(
|
||||
TransactionPolicyId.QUERY_PRIMARY,
|
||||
BUDGET,
|
||||
Optional.of(ReadConsistency.EVENTUAL),
|
||||
Optional.empty()))
|
||||
.isInstanceOf(IllegalArgumentException.class)
|
||||
.hasMessageContaining("QUERY_PRIMARY");
|
||||
}
|
||||
|
||||
@Test
|
||||
void replicaEligibleQueryAcceptsOnlyEventualOrBoundedStaleness() {
|
||||
assertThat(
|
||||
new TransactionRequest(
|
||||
TransactionPolicyId.QUERY_REPLICA_ELIGIBLE,
|
||||
BUDGET,
|
||||
Optional.of(ReadConsistency.BOUNDED_STALENESS),
|
||||
Optional.empty())
|
||||
.readConsistency())
|
||||
.contains(ReadConsistency.BOUNDED_STALENESS);
|
||||
assertThatThrownBy(
|
||||
() ->
|
||||
new TransactionRequest(
|
||||
TransactionPolicyId.QUERY_REPLICA_ELIGIBLE,
|
||||
BUDGET,
|
||||
Optional.of(ReadConsistency.STRONG),
|
||||
Optional.empty()))
|
||||
.isInstanceOf(IllegalArgumentException.class)
|
||||
.hasMessageContaining("QUERY_REPLICA_ELIGIBLE");
|
||||
}
|
||||
|
||||
@Test
|
||||
void readPoliciesRequireAnExplicitConsistency() {
|
||||
assertThatThrownBy(
|
||||
() ->
|
||||
new TransactionRequest(
|
||||
TransactionPolicyId.QUERY_PRIMARY, BUDGET, Optional.empty(), Optional.empty()))
|
||||
.isInstanceOf(IllegalArgumentException.class)
|
||||
.hasMessageContaining("readConsistency");
|
||||
}
|
||||
|
||||
@Test
|
||||
void outboxAppendCanInheritTheOuterOperationIdentity() {
|
||||
TransactionRequest request =
|
||||
new TransactionRequest(
|
||||
TransactionPolicyId.OUTBOX_APPEND, BUDGET, Optional.empty(), Optional.empty());
|
||||
|
||||
assertThat(request.operationId()).isEmpty();
|
||||
}
|
||||
}
|
||||
+58
@@ -0,0 +1,58 @@
|
||||
package dev.caskeleton.application.transaction;
|
||||
|
||||
import static org.assertj.core.api.Assertions.assertThat;
|
||||
import static org.assertj.core.api.Assertions.assertThatThrownBy;
|
||||
|
||||
import java.util.Optional;
|
||||
import org.junit.jupiter.api.Test;
|
||||
|
||||
class TransactionResultTest {
|
||||
|
||||
private static final OperationId OPERATION_ID = new OperationId("operation-42");
|
||||
|
||||
@Test
|
||||
void representsCommittedAndParticipatingResultsWithoutConflatingThem() {
|
||||
TransactionResult<String> committed =
|
||||
new TransactionResult.Committed<>("value", Optional.of(OPERATION_ID));
|
||||
TransactionResult<String> participating = new TransactionResult.Participating<>("value");
|
||||
|
||||
assertThat(committed.outcome()).isEqualTo(TransactionOutcome.COMMITTED);
|
||||
assertThat(participating.outcome()).isEqualTo(TransactionOutcome.PARTICIPATING_PENDING_OUTER);
|
||||
}
|
||||
|
||||
@Test
|
||||
void indeterminateOutcomeCarriesOnlyOptionalSafeReconciliationData() {
|
||||
ReconciliationReference reference =
|
||||
new ReconciliationReference("operation-ledger:operation-42");
|
||||
TransactionResult<String> result =
|
||||
new TransactionResult.Indeterminate<>(
|
||||
Optional.of(OPERATION_ID), TransactionPhase.COMMIT_REQUESTED, Optional.of(reference));
|
||||
|
||||
assertThat(result.outcome()).isEqualTo(TransactionOutcome.INDETERMINATE);
|
||||
assertThat(((TransactionResult.Indeterminate<String>) result).reconciliationReference())
|
||||
.contains(reference);
|
||||
}
|
||||
|
||||
@Test
|
||||
void determinateRollbackAndPostCommitFailureKeepDifferentOutcomes() {
|
||||
RuntimeException rollback = new IllegalStateException("rolled back");
|
||||
RuntimeException cleanup = new IllegalStateException("cleanup failed");
|
||||
|
||||
TransactionResult<String> rolledBack = new TransactionResult.DeterminateRollback<>(rollback);
|
||||
TransactionResult<String> committedWithFailure =
|
||||
new TransactionResult.CommittedWithPostCommitFailure<>(
|
||||
"value", Optional.of(OPERATION_ID), cleanup);
|
||||
|
||||
assertThat(rolledBack.outcome()).isEqualTo(TransactionOutcome.DETERMINATE_ROLLBACK);
|
||||
assertThat(committedWithFailure.outcome())
|
||||
.isEqualTo(TransactionOutcome.COMMITTED_WITH_POST_COMMIT_FAILURE);
|
||||
}
|
||||
|
||||
@Test
|
||||
void reconciliationReferenceRejectsUnsafeOrUnboundedText() {
|
||||
assertThatThrownBy(() -> new ReconciliationReference("raw\nsql"))
|
||||
.isInstanceOf(IllegalArgumentException.class);
|
||||
assertThatThrownBy(() -> new ReconciliationReference("a".repeat(257)))
|
||||
.isInstanceOf(IllegalArgumentException.class);
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user