refactor: adapter 구현중..
This commit is contained in:
DongHyeonka
+18
-1
@@ -13,7 +13,24 @@ public class IdempotencyInFlightException extends RuntimeException {
|
||||
private final transient IdempotencyScope scope;
|
||||
|
||||
public IdempotencyInFlightException(IdempotencyScope scope) {
|
||||
super("idempotent request still in flight: " + (scope == null ? "<null>" : scope.storageKey()));
|
||||
this(scope == null ? "<null>" : scope.storageKey(), scope);
|
||||
}
|
||||
|
||||
/**
|
||||
* Creates the exception for the owner-safe lifecycle, which never holds the raw scope: the
|
||||
* principal and the client key are digested before they reach the store, so the opaque digest
|
||||
* diagnostic is the only identity available. {@link #scope()} is {@code null} on this path, as it
|
||||
* already is for any deserialized instance — the field is {@code transient}.
|
||||
*
|
||||
* @param scopeDigestDiagnostic opaque scope-digest diagnostic, for logs only
|
||||
* @return the exception
|
||||
*/
|
||||
public static IdempotencyInFlightException forScopeDigest(String scopeDigestDiagnostic) {
|
||||
return new IdempotencyInFlightException(scopeDigestDiagnostic, null);
|
||||
}
|
||||
|
||||
private IdempotencyInFlightException(String scopeDiagnostic, IdempotencyScope scope) {
|
||||
super("idempotent request still in flight: " + scopeDiagnostic);
|
||||
this.scope = scope;
|
||||
}
|
||||
|
||||
|
||||
+18
-3
@@ -13,9 +13,24 @@ public class IdempotencyRequestMismatchException extends RuntimeException {
|
||||
private final transient IdempotencyScope scope;
|
||||
|
||||
public IdempotencyRequestMismatchException(IdempotencyScope scope) {
|
||||
super(
|
||||
"idempotency key reused with a different request body: "
|
||||
+ (scope == null ? "<null>" : scope.storageKey()));
|
||||
this(scope == null ? "<null>" : scope.storageKey(), scope);
|
||||
}
|
||||
|
||||
/**
|
||||
* Creates the exception for the owner-safe lifecycle, which never holds the raw scope: the
|
||||
* principal and the client key are digested before they reach the store, so the opaque digest
|
||||
* diagnostic is the only identity available. {@link #scope()} is {@code null} on this path, as it
|
||||
* already is for any deserialized instance — the field is {@code transient}.
|
||||
*
|
||||
* @param scopeDigestDiagnostic opaque scope-digest diagnostic, for logs only
|
||||
* @return the exception
|
||||
*/
|
||||
public static IdempotencyRequestMismatchException forScopeDigest(String scopeDigestDiagnostic) {
|
||||
return new IdempotencyRequestMismatchException(scopeDigestDiagnostic, null);
|
||||
}
|
||||
|
||||
private IdempotencyRequestMismatchException(String scopeDiagnostic, IdempotencyScope scope) {
|
||||
super("idempotency key reused with a different request body: " + scopeDiagnostic);
|
||||
this.scope = scope;
|
||||
}
|
||||
|
||||
|
||||
+328
@@ -0,0 +1,328 @@
|
||||
package dev.caskeleton.application.idempotency.v2;
|
||||
|
||||
import dev.caskeleton.application.idempotency.IdempotencyInFlightException;
|
||||
import dev.caskeleton.application.idempotency.IdempotencyRecoveryRequiredException;
|
||||
import dev.caskeleton.application.idempotency.IdempotencyRequestMismatchException;
|
||||
import dev.caskeleton.application.idempotency.IdempotencyUnavailableException;
|
||||
import dev.caskeleton.application.idempotency.IdempotentAction;
|
||||
import dev.caskeleton.application.idempotency.IdempotentResponseCodec;
|
||||
import dev.caskeleton.application.idempotency.RequestFingerprint;
|
||||
import dev.caskeleton.application.idempotency.StoredResponse;
|
||||
import dev.caskeleton.application.transaction.OperationId;
|
||||
import java.time.Duration;
|
||||
import java.util.Objects;
|
||||
import java.util.Optional;
|
||||
|
||||
/**
|
||||
* Owner-safe request-replay lifecycle, over the contract the providers actually implement.
|
||||
*
|
||||
* <p>There were two V2 store contracts with the same name in neighbouring packages, and this
|
||||
* orchestration was written against the one nothing implements. Both owner-safe stores — PostgreSQL
|
||||
* and Redis — implement {@link IdempotencyStorePortV2} here, so selecting a V2 provider required a
|
||||
* bean that could not exist and the executor could never be composed at all.
|
||||
*
|
||||
* <p>Porting it was not a rename. This contract threads the owner handle through every mutation: a
|
||||
* confirmed transition returns the owner with its state revision advanced, and the next call must
|
||||
* present <em>that</em> handle. Re-using the claim's original owner would present a stale revision,
|
||||
* which the store is built to refuse — that refusal is the whole point of the compare-and-set, so a
|
||||
* caller that defeats it has an owner-safe store and no owner safety.
|
||||
*
|
||||
* <p>The action runs only after a confirmed start. This preserves request-replay evidence; it does
|
||||
* not create a cross-store exactly-once boundary, and nothing here should be read as claiming one.
|
||||
*/
|
||||
public final class IdempotencyExecutorV2 {
|
||||
|
||||
private final IdempotencyStorePortV2 store;
|
||||
private final Duration processingLeaseTtl;
|
||||
private final Duration replayTtl;
|
||||
private final Duration failureRetention;
|
||||
private final String responseCodecId;
|
||||
private final int policyRevision;
|
||||
|
||||
/**
|
||||
* Creates the executor.
|
||||
*
|
||||
* @param store the owner-safe store
|
||||
* @param processingLeaseTtl how long a claim holds the record before another caller may take over
|
||||
* @param replayTtl how long a completed response is replayable
|
||||
* @param failureRetention how long a failed attempt's evidence is kept
|
||||
* @param responseCodecId the identifier of the codec the stored payload was written with
|
||||
* @param policyRevision the replay policy revision this deployment enforces
|
||||
*/
|
||||
public IdempotencyExecutorV2(
|
||||
IdempotencyStorePortV2 store,
|
||||
Duration processingLeaseTtl,
|
||||
Duration replayTtl,
|
||||
Duration failureRetention,
|
||||
String responseCodecId,
|
||||
int policyRevision) {
|
||||
this.store = Objects.requireNonNull(store, "store must be non-null");
|
||||
this.processingLeaseTtl =
|
||||
Objects.requireNonNull(processingLeaseTtl, "processing lease TTL must be non-null");
|
||||
this.replayTtl = Objects.requireNonNull(replayTtl, "replay TTL must be non-null");
|
||||
this.failureRetention =
|
||||
Objects.requireNonNull(failureRetention, "failure retention must be non-null");
|
||||
this.responseCodecId =
|
||||
Objects.requireNonNull(responseCodecId, "response codec identifier must be non-null");
|
||||
this.policyRevision = policyRevision;
|
||||
if (failureRetention.isZero() || failureRetention.isNegative()) {
|
||||
throw new IllegalArgumentException("the failure retention must be positive");
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Mints the attempt a caller retains across a lost response.
|
||||
*
|
||||
* @param operationId the caller's operation identifier
|
||||
* @return the claim attempt
|
||||
*/
|
||||
public IdempotencyClaimAttempt newAttempt(OperationId operationId) {
|
||||
return store.newClaimAttempt(operationId);
|
||||
}
|
||||
|
||||
/**
|
||||
* Claims the scope, runs the action once, and stores its response for replay.
|
||||
*
|
||||
* @param scope the digested scope; the raw client key never reaches this class
|
||||
* @param fingerprint the request fingerprint a replay is checked against
|
||||
* @param attempt the retained claim attempt
|
||||
* @param action the action to run at most once
|
||||
* @param codec serialises and deserialises the action's result
|
||||
* @param <R> the action's result type
|
||||
* @return the action's result, or the replayed one
|
||||
*/
|
||||
public <R> R execute(
|
||||
IdempotencyScopeDigest scope,
|
||||
RequestFingerprint fingerprint,
|
||||
IdempotencyClaimAttempt attempt,
|
||||
IdempotentAction<R> action,
|
||||
IdempotentResponseCodec<R> codec) {
|
||||
Objects.requireNonNull(action, "action must be non-null");
|
||||
Objects.requireNonNull(codec, "codec must be non-null");
|
||||
IdempotencyClaimRequest request =
|
||||
new IdempotencyClaimRequest(
|
||||
scope,
|
||||
fingerprint,
|
||||
attempt,
|
||||
processingLeaseTtl,
|
||||
replayTtl,
|
||||
responseCodecId,
|
||||
policyRevision);
|
||||
|
||||
return switch (store.claim(request)) {
|
||||
case IdempotencyClaimOutcome.CompletedReplay replay ->
|
||||
codec.deserialize(replay.response().payload());
|
||||
case IdempotencyClaimOutcome.FingerprintMismatch ignored ->
|
||||
throw IdempotencyRequestMismatchException.forScopeDigest(diagnostic(scope));
|
||||
case IdempotencyClaimOutcome.InProgress ignored ->
|
||||
throw IdempotencyInFlightException.forScopeDigest(diagnostic(scope));
|
||||
case IdempotencyClaimOutcome.RecoveryRequired ignored ->
|
||||
throw recovery("claim requires reconciliation");
|
||||
case IdempotencyClaimOutcome.OwnerOperationConflict ignored ->
|
||||
throw recovery("claim requires reconciliation");
|
||||
case IdempotencyClaimOutcome.Unavailable ignored ->
|
||||
throw new IdempotencyUnavailableException();
|
||||
// The response was lost, not the attempt. The retained attempt is what makes the record
|
||||
// findable, so the only safe move is to ask the store what actually happened.
|
||||
case IdempotencyClaimOutcome.Indeterminate ignored -> reconcile(request, action, codec);
|
||||
case IdempotencyClaimOutcome.Acquired acquired ->
|
||||
startAndRun(request, acquired.owner(), action, codec, false);
|
||||
case IdempotencyClaimOutcome.ReplayedAcquire replayed ->
|
||||
startAndRun(request, replayed.owner(), action, codec, false);
|
||||
case IdempotencyClaimOutcome.TakenOverClaimed takenOver ->
|
||||
startAndRun(request, takenOver.owner(), action, codec, false);
|
||||
};
|
||||
}
|
||||
|
||||
private <R> R reconcile(
|
||||
IdempotencyClaimRequest request,
|
||||
IdempotentAction<R> action,
|
||||
IdempotentResponseCodec<R> codec) {
|
||||
IdempotencyInspection inspection = inspect(request);
|
||||
return switch (inspection.outcome()) {
|
||||
case CLAIMED_SAME_OPERATION ->
|
||||
startAndRun(request, requireOwner(inspection), action, codec, false);
|
||||
case EXECUTING_SAME_OPERATION -> runStarted(request, requireOwner(inspection), action, codec);
|
||||
case COMPLETED_REPLAY -> codec.deserialize(requireResponse(inspection).payload());
|
||||
case FINGERPRINT_MISMATCH ->
|
||||
throw IdempotencyRequestMismatchException.forScopeDigest(diagnostic(request.scope()));
|
||||
case IN_PROGRESS_OTHER ->
|
||||
throw IdempotencyInFlightException.forScopeDigest(diagnostic(request.scope()));
|
||||
case UNAVAILABLE -> throw new IdempotencyUnavailableException();
|
||||
// ABSENT, FAILED_RETRYABLE, ABANDONED and OPERATION_CONFLICT all mean the same thing here:
|
||||
// this caller cannot prove what happened to its own attempt, and guessing is the one thing an
|
||||
// owner-safe store exists to prevent.
|
||||
default -> throw recovery("an indeterminate claim could not be resumed safely");
|
||||
};
|
||||
}
|
||||
|
||||
private <R> R startAndRun(
|
||||
IdempotencyClaimRequest request,
|
||||
IdempotencyOwner owner,
|
||||
IdempotentAction<R> action,
|
||||
IdempotentResponseCodec<R> codec,
|
||||
boolean alreadyRetried) {
|
||||
IdempotencyMutationResult<IdempotencyStartOutcome> started =
|
||||
store.markExecutionStarted(owner, request.claimAttempt().operationId());
|
||||
return switch (started.outcome()) {
|
||||
// The owner handle from the transition, not the one from the claim: the state revision has
|
||||
// advanced and the next compare-and-set is against the new one.
|
||||
case STARTED, ALREADY_STARTED_SAME_OPERATION ->
|
||||
runStarted(request, advanced(started, owner), action, codec);
|
||||
case UNAVAILABLE -> throw new IdempotencyUnavailableException();
|
||||
case INDETERMINATE -> {
|
||||
if (alreadyRetried) {
|
||||
throw recovery("execution start stayed indeterminate after reconciliation");
|
||||
}
|
||||
yield resumeAfterIndeterminateStart(request, action, codec);
|
||||
}
|
||||
default -> throw recovery("execution start was not confirmed for this exact operation");
|
||||
};
|
||||
}
|
||||
|
||||
private <R> R resumeAfterIndeterminateStart(
|
||||
IdempotencyClaimRequest request,
|
||||
IdempotentAction<R> action,
|
||||
IdempotentResponseCodec<R> codec) {
|
||||
IdempotencyInspection inspection = inspect(request);
|
||||
return switch (inspection.outcome()) {
|
||||
case CLAIMED_SAME_OPERATION ->
|
||||
startAndRun(request, requireOwner(inspection), action, codec, true);
|
||||
case EXECUTING_SAME_OPERATION -> runStarted(request, requireOwner(inspection), action, codec);
|
||||
case COMPLETED_REPLAY -> codec.deserialize(requireResponse(inspection).payload());
|
||||
case UNAVAILABLE -> throw new IdempotencyUnavailableException();
|
||||
default -> throw recovery("execution start is indeterminate");
|
||||
};
|
||||
}
|
||||
|
||||
private <R> R runStarted(
|
||||
IdempotencyClaimRequest request,
|
||||
IdempotencyOwner owner,
|
||||
IdempotentAction<R> action,
|
||||
IdempotentResponseCodec<R> codec) {
|
||||
OperationId operationId = request.claimAttempt().operationId();
|
||||
IdempotentAction.Outcome<R> outcome;
|
||||
try {
|
||||
outcome = Objects.requireNonNull(action.run(), "action outcome must be non-null");
|
||||
} catch (RuntimeException failure) {
|
||||
// The action threw without saying whether it had an effect, so the record must not say
|
||||
// "retryable": that would invite a second execution of something that may already have run.
|
||||
preserveUnknown(owner, operationId);
|
||||
throw failure;
|
||||
}
|
||||
return switch (outcome) {
|
||||
case IdempotentAction.Outcome.Success<R> success ->
|
||||
complete(request, owner, success.result(), codec, false);
|
||||
case IdempotentAction.Outcome.RetryableNoEffect<R> retryable -> {
|
||||
store.markFailed(
|
||||
owner,
|
||||
IdempotencyFailureDisposition.NO_EFFECT_RETRYABLE,
|
||||
failureRetention,
|
||||
operationId);
|
||||
throw retryable.failure();
|
||||
}
|
||||
case IdempotentAction.Outcome.EffectUnknown<R> unknown -> {
|
||||
preserveUnknown(owner, operationId);
|
||||
throw unknown.failure();
|
||||
}
|
||||
};
|
||||
}
|
||||
|
||||
private <R> R complete(
|
||||
IdempotencyClaimRequest request,
|
||||
IdempotencyOwner owner,
|
||||
R result,
|
||||
IdempotentResponseCodec<R> codec,
|
||||
boolean alreadyRetried) {
|
||||
StoredResponse response = new StoredResponse(codec.serialize(result));
|
||||
return switch (store.complete(
|
||||
owner, response, replayTtl, request.claimAttempt().operationId())) {
|
||||
case COMPLETED, ALREADY_COMPLETED_SAME_RESULT -> result;
|
||||
case UNAVAILABLE -> throw new IdempotencyUnavailableException();
|
||||
case INDETERMINATE -> reconcileCompletion(request, result, codec, alreadyRetried);
|
||||
default -> throw recovery("completion was not confirmed");
|
||||
};
|
||||
}
|
||||
|
||||
/**
|
||||
* Resolves a completion whose response was lost.
|
||||
*
|
||||
* <p>Deliberately does not carry the caller's owner handle forward: after an indeterminate
|
||||
* completion the caller's handle may already be stale, and the only handle worth presenting is
|
||||
* the one the store reports now.
|
||||
*/
|
||||
private <R> R reconcileCompletion(
|
||||
IdempotencyClaimRequest request,
|
||||
R result,
|
||||
IdempotentResponseCodec<R> codec,
|
||||
boolean alreadyRetried) {
|
||||
IdempotencyInspection inspection = inspect(request);
|
||||
return switch (inspection.outcome()) {
|
||||
case COMPLETED_REPLAY -> {
|
||||
StoredResponse stored = requireResponse(inspection);
|
||||
if (stored.payload().equals(codec.serialize(result))) {
|
||||
yield result;
|
||||
}
|
||||
// The stored response is somebody else's answer to this scope. Returning either one would
|
||||
// be asserting a fact this caller cannot establish.
|
||||
throw recovery("the completed response conflicts with the one this caller produced");
|
||||
}
|
||||
case EXECUTING_SAME_OPERATION -> {
|
||||
if (alreadyRetried) {
|
||||
throw recovery("completion stayed indeterminate after reconciliation");
|
||||
}
|
||||
yield complete(request, requireOwner(inspection), result, codec, true);
|
||||
}
|
||||
case UNAVAILABLE -> throw new IdempotencyUnavailableException();
|
||||
default -> throw recovery("the completion response is indeterminate and was not reconciled");
|
||||
};
|
||||
}
|
||||
|
||||
private void preserveUnknown(IdempotencyOwner owner, OperationId operationId) {
|
||||
store.markFailed(
|
||||
owner,
|
||||
IdempotencyFailureDisposition.EFFECT_UNKNOWN_ABANDONED,
|
||||
failureRetention,
|
||||
operationId);
|
||||
}
|
||||
|
||||
private IdempotencyInspection inspect(IdempotencyClaimRequest request) {
|
||||
return store.inspect(
|
||||
new IdempotencyInspectionRequest(
|
||||
request.scope(), request.requestFingerprint(), request.claimAttempt()));
|
||||
}
|
||||
|
||||
private static IdempotencyOwner advanced(
|
||||
IdempotencyMutationResult<IdempotencyStartOutcome> transition, IdempotencyOwner fallback) {
|
||||
return transition.owner().orElse(fallback);
|
||||
}
|
||||
|
||||
private static IdempotencyOwner requireOwner(IdempotencyInspection inspection) {
|
||||
Optional<IdempotencyOwner> owner = inspection.owner();
|
||||
return owner.orElseThrow(
|
||||
() ->
|
||||
recovery(
|
||||
"the store reported "
|
||||
+ inspection.outcome()
|
||||
+ " without the owner handle it needs"));
|
||||
}
|
||||
|
||||
private static StoredResponse requireResponse(IdempotencyInspection inspection) {
|
||||
return inspection
|
||||
.response()
|
||||
.orElseThrow(() -> recovery("the store reported a completed replay with no response"));
|
||||
}
|
||||
|
||||
/**
|
||||
* Renders the scope for a diagnostic. Every component is already opaque — the raw client key and
|
||||
* principal were digested before the scope was constructed — so this is safe to log, and it is
|
||||
* the only scope identity this lifecycle ever holds.
|
||||
*/
|
||||
private static String diagnostic(IdempotencyScopeDigest scope) {
|
||||
return scope.operationCode() + "::v" + scope.keyDigestVersion() + "::" + scope.digest();
|
||||
}
|
||||
|
||||
private static IdempotencyRecoveryRequiredException recovery(String message) {
|
||||
return new IdempotencyRecoveryRequiredException(message);
|
||||
}
|
||||
}
|
||||
+343
@@ -0,0 +1,343 @@
|
||||
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.IdempotentAction;
|
||||
import dev.caskeleton.application.idempotency.IdempotentResponseCodec;
|
||||
import dev.caskeleton.application.idempotency.RequestFingerprint;
|
||||
import dev.caskeleton.application.idempotency.StoredResponse;
|
||||
import dev.caskeleton.application.transaction.OperationId;
|
||||
import java.time.Duration;
|
||||
import java.time.Instant;
|
||||
import java.util.Optional;
|
||||
import java.util.concurrent.atomic.AtomicBoolean;
|
||||
import org.junit.jupiter.api.DisplayName;
|
||||
import org.junit.jupiter.api.Test;
|
||||
|
||||
/**
|
||||
* The request-replay lifecycle, over the contract the stores implement.
|
||||
*
|
||||
* <p>Carried over from a test of the same name that exercised the duplicate V2 contract in the
|
||||
* parent package — the one nothing implements. The behaviours below are the ones worth keeping: the
|
||||
* action runs at most once, a lost response is reconciled rather than re-run, and an outcome the
|
||||
* caller cannot classify is never recorded as "no effect".
|
||||
*
|
||||
* <p>One behaviour is new, because the contract is: every confirmed transition hands back the owner
|
||||
* with its state revision advanced, and the next call must present that handle rather than the one
|
||||
* the claim returned.
|
||||
*/
|
||||
class IdempotencyExecutorV2Test {
|
||||
|
||||
private static final String OWNER_TOKEN = "b".repeat(64);
|
||||
private static final OperationId OPERATION = new OperationId("operation-aaaaaaaaaa");
|
||||
private static final IdempotencyScopeDigest SCOPE =
|
||||
new IdempotencyScopeDigest("a".repeat(64), 1, "CREATE_WORKLOG");
|
||||
private static final RequestFingerprint FINGERPRINT = new RequestFingerprint("c".repeat(64));
|
||||
private static final IdempotencyClaimAttempt ATTEMPT =
|
||||
new IdempotencyClaimAttempt(OWNER_TOKEN, OPERATION);
|
||||
private static final Instant LEASE_UNTIL = Instant.parse("2026-07-29T12:00:30Z");
|
||||
|
||||
private static IdempotencyOwner owner(long stateRevision) {
|
||||
return new IdempotencyOwner(SCOPE, OWNER_TOKEN, 1, stateRevision, OPERATION);
|
||||
}
|
||||
|
||||
@Test
|
||||
@DisplayName("the action runs only after a confirmed start, and then completes")
|
||||
void actionRunsOnlyAfterAConfirmedStart() {
|
||||
FakeStore store = new FakeStore();
|
||||
store.claim = new IdempotencyClaimOutcome.Acquired(owner(0), LEASE_UNTIL);
|
||||
store.start = IdempotencyStartOutcome.STARTED;
|
||||
AtomicBoolean ran = new AtomicBoolean();
|
||||
|
||||
String result =
|
||||
executor(store)
|
||||
.execute(
|
||||
SCOPE,
|
||||
FINGERPRINT,
|
||||
ATTEMPT,
|
||||
() -> {
|
||||
ran.set(true);
|
||||
return new IdempotentAction.Outcome.Success<>("created");
|
||||
},
|
||||
codec());
|
||||
|
||||
assertThat(result).isEqualTo("created");
|
||||
assertThat(ran).isTrue();
|
||||
assertThat(store.completeCalls).isEqualTo(1);
|
||||
assertThat(store.failedCalls).isZero();
|
||||
}
|
||||
|
||||
@Test
|
||||
@DisplayName("each mutation presents the owner handle the previous one returned")
|
||||
void theAdvancedOwnerHandleIsCarriedForward() {
|
||||
// The contract's whole point. The store advances the state revision on a confirmed transition
|
||||
// and refuses a stale one; a caller that kept presenting the claim's original handle would be
|
||||
// refused by its own store, so an owner-safe store would buy nothing.
|
||||
FakeStore store = new FakeStore();
|
||||
store.claim = new IdempotencyClaimOutcome.Acquired(owner(0), LEASE_UNTIL);
|
||||
store.start = IdempotencyStartOutcome.STARTED;
|
||||
store.startOwner = owner(1);
|
||||
|
||||
executor(store)
|
||||
.execute(
|
||||
SCOPE,
|
||||
FINGERPRINT,
|
||||
ATTEMPT,
|
||||
() -> new IdempotentAction.Outcome.Success<>("created"),
|
||||
codec());
|
||||
|
||||
assertThat(store.completeOwner.stateRevision())
|
||||
.as("complete must present the revision the start returned, not the claim's")
|
||||
.isEqualTo(1);
|
||||
}
|
||||
|
||||
@Test
|
||||
@DisplayName("an indeterminate claim is reconciled rather than re-run")
|
||||
void anIndeterminateClaimIsReconciled() {
|
||||
FakeStore store = new FakeStore();
|
||||
store.claim = new IdempotencyClaimOutcome.Indeterminate(OPERATION);
|
||||
store.inspection =
|
||||
new IdempotencyInspection(
|
||||
IdempotencyInspectionOutcome.EXECUTING_SAME_OPERATION,
|
||||
Optional.of(owner(1)),
|
||||
Optional.of(LEASE_UNTIL),
|
||||
Optional.empty(),
|
||||
Optional.empty());
|
||||
AtomicBoolean ran = new AtomicBoolean();
|
||||
|
||||
assertThat(
|
||||
executor(store)
|
||||
.execute(
|
||||
SCOPE,
|
||||
FINGERPRINT,
|
||||
ATTEMPT,
|
||||
() -> {
|
||||
ran.set(true);
|
||||
return new IdempotentAction.Outcome.Success<>("created");
|
||||
},
|
||||
codec()))
|
||||
.isEqualTo("created");
|
||||
assertThat(ran).isTrue();
|
||||
assertThat(store.startCalls)
|
||||
.as("the record already says EXECUTING; starting it again would be a second transition")
|
||||
.isZero();
|
||||
}
|
||||
|
||||
@Test
|
||||
@DisplayName("a lost completion is inspected without running the action a second time")
|
||||
void aLostCompletionIsInspectedNotRepeated() {
|
||||
FakeStore store = startedStore();
|
||||
store.complete = IdempotencyCompleteOutcome.INDETERMINATE;
|
||||
store.inspectionAfterComplete =
|
||||
new IdempotencyInspection(
|
||||
IdempotencyInspectionOutcome.COMPLETED_REPLAY,
|
||||
Optional.empty(),
|
||||
Optional.empty(),
|
||||
Optional.of(new StoredResponse("created")),
|
||||
Optional.of(Instant.parse("2026-07-29T13:00:00Z")));
|
||||
int[] actionCalls = {0};
|
||||
|
||||
assertThat(
|
||||
executor(store)
|
||||
.execute(
|
||||
SCOPE,
|
||||
FINGERPRINT,
|
||||
ATTEMPT,
|
||||
() -> {
|
||||
actionCalls[0]++;
|
||||
return new IdempotentAction.Outcome.Success<>("created");
|
||||
},
|
||||
codec()))
|
||||
.isEqualTo("created");
|
||||
assertThat(actionCalls[0]).isEqualTo(1);
|
||||
assertThat(store.completeCalls).isEqualTo(1);
|
||||
}
|
||||
|
||||
@Test
|
||||
@DisplayName("a replayed response that disagrees with this caller's is a recovery, not a result")
|
||||
void aConflictingReplayIsNotReturned() {
|
||||
FakeStore store = startedStore();
|
||||
store.complete = IdempotencyCompleteOutcome.INDETERMINATE;
|
||||
store.inspectionAfterComplete =
|
||||
new IdempotencyInspection(
|
||||
IdempotencyInspectionOutcome.COMPLETED_REPLAY,
|
||||
Optional.empty(),
|
||||
Optional.empty(),
|
||||
Optional.of(new StoredResponse("somebody-elses-answer")),
|
||||
Optional.of(Instant.parse("2026-07-29T13:00:00Z")));
|
||||
|
||||
assertThatThrownBy(
|
||||
() ->
|
||||
executor(store)
|
||||
.execute(
|
||||
SCOPE,
|
||||
FINGERPRINT,
|
||||
ATTEMPT,
|
||||
() -> new IdempotentAction.Outcome.Success<>("created"),
|
||||
codec()))
|
||||
.hasMessageContaining("conflicts");
|
||||
}
|
||||
|
||||
@Test
|
||||
@DisplayName("an unclassified throw and an unknown effect are never recorded as no-effect")
|
||||
void unknownEffectsAreNeverDiscarded() {
|
||||
FakeStore thrown = startedStore();
|
||||
RuntimeException unclassified = new IllegalStateException("unknown effect");
|
||||
|
||||
assertThatThrownBy(
|
||||
() ->
|
||||
executor(thrown)
|
||||
.execute(
|
||||
SCOPE,
|
||||
FINGERPRINT,
|
||||
ATTEMPT,
|
||||
() -> {
|
||||
throw unclassified;
|
||||
},
|
||||
codec()))
|
||||
.isSameAs(unclassified);
|
||||
assertThat(thrown.lastDisposition)
|
||||
.isEqualTo(IdempotencyFailureDisposition.EFFECT_UNKNOWN_ABANDONED);
|
||||
|
||||
FakeStore declared = startedStore();
|
||||
RuntimeException classified = new IllegalArgumentException("provider response unknown");
|
||||
assertThatThrownBy(
|
||||
() ->
|
||||
executor(declared)
|
||||
.execute(
|
||||
SCOPE,
|
||||
FINGERPRINT,
|
||||
ATTEMPT,
|
||||
() -> new IdempotentAction.Outcome.EffectUnknown<>(classified),
|
||||
codec()))
|
||||
.isSameAs(classified);
|
||||
assertThat(declared.lastDisposition)
|
||||
.isEqualTo(IdempotencyFailureDisposition.EFFECT_UNKNOWN_ABANDONED);
|
||||
}
|
||||
|
||||
@Test
|
||||
@DisplayName("only an explicitly effect-free failure is recorded as retryable")
|
||||
void onlyDeclaredNoEffectIsRetryable() {
|
||||
FakeStore store = startedStore();
|
||||
RuntimeException failure = new IllegalArgumentException("validation");
|
||||
|
||||
assertThatThrownBy(
|
||||
() ->
|
||||
executor(store)
|
||||
.execute(
|
||||
SCOPE,
|
||||
FINGERPRINT,
|
||||
ATTEMPT,
|
||||
() -> new IdempotentAction.Outcome.RetryableNoEffect<>(failure),
|
||||
codec()))
|
||||
.isSameAs(failure);
|
||||
assertThat(store.lastDisposition).isEqualTo(IdempotencyFailureDisposition.NO_EFFECT_RETRYABLE);
|
||||
}
|
||||
|
||||
private static FakeStore startedStore() {
|
||||
FakeStore store = new FakeStore();
|
||||
store.claim = new IdempotencyClaimOutcome.Acquired(owner(0), LEASE_UNTIL);
|
||||
store.start = IdempotencyStartOutcome.STARTED;
|
||||
return store;
|
||||
}
|
||||
|
||||
private static IdempotencyExecutorV2 executor(FakeStore store) {
|
||||
return new IdempotencyExecutorV2(
|
||||
store, Duration.ofSeconds(30), Duration.ofHours(1), Duration.ofHours(1), "json-v2", 1);
|
||||
}
|
||||
|
||||
private static IdempotentResponseCodec<String> codec() {
|
||||
return new IdempotentResponseCodec<>() {
|
||||
@Override
|
||||
public String serialize(String result) {
|
||||
return result;
|
||||
}
|
||||
|
||||
@Override
|
||||
public String deserialize(String payload) {
|
||||
return payload;
|
||||
}
|
||||
};
|
||||
}
|
||||
|
||||
private static final class FakeStore implements IdempotencyStorePortV2 {
|
||||
|
||||
private IdempotencyClaimOutcome claim;
|
||||
private IdempotencyStartOutcome start;
|
||||
private IdempotencyOwner startOwner;
|
||||
private IdempotencyCompleteOutcome complete = IdempotencyCompleteOutcome.COMPLETED;
|
||||
private IdempotencyInspection inspection =
|
||||
IdempotencyInspection.outcome(IdempotencyInspectionOutcome.UNAVAILABLE);
|
||||
private IdempotencyInspection inspectionAfterComplete;
|
||||
private int startCalls;
|
||||
private int completeCalls;
|
||||
private int failedCalls;
|
||||
private IdempotencyOwner completeOwner;
|
||||
private IdempotencyFailureDisposition lastDisposition;
|
||||
|
||||
@Override
|
||||
public IdempotencyClaimAttempt newClaimAttempt(OperationId operationId) {
|
||||
return new IdempotencyClaimAttempt(OWNER_TOKEN, operationId);
|
||||
}
|
||||
|
||||
@Override
|
||||
public IdempotencyClaimOutcome claim(IdempotencyClaimRequest request) {
|
||||
return claim;
|
||||
}
|
||||
|
||||
@Override
|
||||
public IdempotencyMutationResult<IdempotencyStartOutcome> markExecutionStarted(
|
||||
IdempotencyOwner owner, OperationId operationId) {
|
||||
startCalls++;
|
||||
IdempotencyOwner advanced =
|
||||
start.carriesOwner() ? (startOwner == null ? owner : startOwner) : null;
|
||||
return new IdempotencyMutationResult<>(
|
||||
start, advanced, IdempotencyStartOutcome::carriesOwner);
|
||||
}
|
||||
|
||||
@Override
|
||||
public IdempotencyMutationResult<IdempotencyRenewOutcome> renew(
|
||||
IdempotencyOwner owner, Duration processingLeaseTtl, OperationId operationId) {
|
||||
throw new UnsupportedOperationException("the executor does not renew");
|
||||
}
|
||||
|
||||
@Override
|
||||
public IdempotencyCompleteOutcome complete(
|
||||
IdempotencyOwner owner,
|
||||
StoredResponse response,
|
||||
Duration replayTtl,
|
||||
OperationId operationId) {
|
||||
completeCalls++;
|
||||
completeOwner = owner;
|
||||
if (inspectionAfterComplete != null) {
|
||||
inspection = inspectionAfterComplete;
|
||||
}
|
||||
return complete;
|
||||
}
|
||||
|
||||
@Override
|
||||
public IdempotencyFailOutcome markFailed(
|
||||
IdempotencyOwner owner,
|
||||
IdempotencyFailureDisposition disposition,
|
||||
Duration retention,
|
||||
OperationId operationId) {
|
||||
failedCalls++;
|
||||
lastDisposition = disposition;
|
||||
return disposition == IdempotencyFailureDisposition.NO_EFFECT_RETRYABLE
|
||||
? IdempotencyFailOutcome.MARKED_RETRYABLE
|
||||
: IdempotencyFailOutcome.MARKED_ABANDONED;
|
||||
}
|
||||
|
||||
@Override
|
||||
public IdempotencyReleaseOutcome releaseBeforeExecution(
|
||||
IdempotencyOwner owner, OperationId operationId) {
|
||||
throw new UnsupportedOperationException("the executor never releases before execution");
|
||||
}
|
||||
|
||||
@Override
|
||||
public IdempotencyInspection inspect(IdempotencyInspectionRequest request) {
|
||||
return inspection;
|
||||
}
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user