feat: add resumable idempotent experiment actions
chenyuan
2 weeks ago
parent
ae8ce2819c
commit
dcd01e57be
@ -0,0 +1,30 @@
|
||||
package com.yau.digitalrmb.training.attempt.application;
|
||||
|
||||
import com.yau.digitalrmb.training.attempt.domain.ActionStatus;
|
||||
|
||||
public final class ActionOutcome<T> {
|
||||
private final T output;
|
||||
private final ActionStatus status;
|
||||
private final String nextStepCode;
|
||||
private final String nextActionCode;
|
||||
|
||||
private ActionOutcome(T output, ActionStatus status, String nextStepCode, String nextActionCode) {
|
||||
this.output = output;
|
||||
this.status = status;
|
||||
this.nextStepCode = nextStepCode;
|
||||
this.nextActionCode = nextActionCode;
|
||||
}
|
||||
|
||||
public static <T> ActionOutcome<T> completed(T output) {
|
||||
return new ActionOutcome<T>(output, ActionStatus.COMPLETED, null, null);
|
||||
}
|
||||
|
||||
public static <T> ActionOutcome<T> rolledBack(T output, String nextStepCode, String nextActionCode) {
|
||||
return new ActionOutcome<T>(output, ActionStatus.ROLLED_BACK, nextStepCode, nextActionCode);
|
||||
}
|
||||
|
||||
public T getOutput() { return output; }
|
||||
public ActionStatus getStatus() { return status; }
|
||||
public String getNextStepCode() { return nextStepCode; }
|
||||
public String getNextActionCode() { return nextActionCode; }
|
||||
}
|
||||
@ -0,0 +1,6 @@
|
||||
package com.yau.digitalrmb.training.attempt.application;
|
||||
|
||||
@FunctionalInterface
|
||||
public interface ActionWork<T> {
|
||||
ActionOutcome<T> run();
|
||||
}
|
||||
@ -0,0 +1,10 @@
|
||||
package com.yau.digitalrmb.training.attempt.application;
|
||||
|
||||
import com.yau.digitalrmb.training.attempt.domain.ExperimentAttempt;
|
||||
import com.yau.digitalrmb.training.attempt.domain.ExperimentModule;
|
||||
import com.yau.digitalrmb.training.attempt.domain.ExperimentSubject;
|
||||
|
||||
public interface AttemptCancellationHandler {
|
||||
ExperimentModule module();
|
||||
void cancel(ExperimentAttempt attempt, ExperimentSubject subject);
|
||||
}
|
||||
@ -0,0 +1,166 @@
|
||||
package com.yau.digitalrmb.training.attempt.application;
|
||||
|
||||
import com.fasterxml.jackson.core.JsonProcessingException;
|
||||
import com.fasterxml.jackson.databind.ObjectMapper;
|
||||
import com.yau.digitalrmb.shared.api.ErrorCode;
|
||||
import com.yau.digitalrmb.shared.exception.BusinessException;
|
||||
import com.yau.digitalrmb.training.attempt.domain.ActionStatus;
|
||||
import com.yau.digitalrmb.training.attempt.domain.AttemptStatus;
|
||||
import com.yau.digitalrmb.training.attempt.domain.ExperimentAction;
|
||||
import com.yau.digitalrmb.training.attempt.domain.ExperimentAttempt;
|
||||
import com.yau.digitalrmb.training.attempt.domain.ExperimentAttemptRepository;
|
||||
import com.yau.digitalrmb.training.attempt.domain.ExperimentModule;
|
||||
import com.yau.digitalrmb.training.attempt.domain.ExperimentSubject;
|
||||
import com.yau.digitalrmb.training.attempt.interfaces.dto.ExperimentActionView;
|
||||
import com.yau.digitalrmb.training.attempt.interfaces.dto.ExperimentAttemptView;
|
||||
import org.springframework.stereotype.Service;
|
||||
import org.springframework.transaction.annotation.Transactional;
|
||||
|
||||
import java.time.Instant;
|
||||
import java.util.ArrayList;
|
||||
import java.util.List;
|
||||
import java.util.UUID;
|
||||
|
||||
@Service
|
||||
public class ExperimentAttemptService {
|
||||
private final ExperimentAttemptRepository repository;
|
||||
private final ObjectMapper objectMapper;
|
||||
private final List<AttemptCancellationHandler> cancellationHandlers;
|
||||
|
||||
public ExperimentAttemptService(ExperimentAttemptRepository repository, ObjectMapper objectMapper,
|
||||
List<AttemptCancellationHandler> cancellationHandlers) {
|
||||
this.repository = repository;
|
||||
this.objectMapper = objectMapper;
|
||||
this.cancellationHandlers = cancellationHandlers;
|
||||
}
|
||||
|
||||
@Transactional
|
||||
public ExperimentAttemptView create(ExperimentModule module, ExperimentSubject subject) {
|
||||
return toAttemptView(repository.create(module, subject), new ArrayList<ExperimentAction>());
|
||||
}
|
||||
|
||||
@Transactional(readOnly = true)
|
||||
public ExperimentAttemptView current(ExperimentModule module, ExperimentSubject subject) {
|
||||
ExperimentAttempt attempt = repository.findCurrent(module, subject)
|
||||
.orElseThrow(() -> notFound("未找到进行中的实验"));
|
||||
return toAttemptView(attempt, repository.findActions(attempt.getId()));
|
||||
}
|
||||
|
||||
@Transactional(readOnly = true)
|
||||
public ExperimentAttemptView detail(UUID attemptId, ExperimentSubject subject) {
|
||||
ExperimentAttempt attempt = repository.findById(attemptId, subject)
|
||||
.orElseThrow(() -> notFound("实验不存在或不属于当前用户"));
|
||||
return toAttemptView(attempt, repository.findActions(attempt.getId()));
|
||||
}
|
||||
|
||||
@Transactional
|
||||
public ExperimentAttemptView cancel(UUID attemptId, ExperimentSubject subject) {
|
||||
ExperimentAttempt attempt = repository.findByIdForUpdate(attemptId, subject)
|
||||
.orElseThrow(() -> notFound("实验不存在或不属于当前用户"));
|
||||
for (AttemptCancellationHandler handler : cancellationHandlers) {
|
||||
if (handler.module() == attempt.getModule()) {
|
||||
handler.cancel(attempt, subject);
|
||||
break;
|
||||
}
|
||||
}
|
||||
attempt.cancel();
|
||||
repository.save(attempt);
|
||||
return toAttemptView(attempt, repository.findActions(attempt.getId()));
|
||||
}
|
||||
|
||||
@Transactional
|
||||
public <T> ExperimentActionView<T> execute(UUID attemptId, ExperimentSubject subject,
|
||||
String stepCode, String actionCode,
|
||||
String requestFingerprint, Class<T> outputType,
|
||||
ActionWork<T> work,
|
||||
String nextStepCode, String nextActionCode) {
|
||||
ExperimentAttempt attempt = repository.findByIdForUpdate(attemptId, subject)
|
||||
.orElseThrow(() -> notFound("实验不存在或不属于当前用户"));
|
||||
requireActive(attempt);
|
||||
|
||||
ExperimentAction action = repository.findActionForUpdate(attemptId, stepCode, actionCode).orElse(null);
|
||||
if (action != null && action.getStatus() == ActionStatus.COMPLETED) {
|
||||
if (!action.getRequestFingerprint().equals(requestFingerprint)) {
|
||||
throw validation("该动作已使用不同参数完成，不能覆盖原结果");
|
||||
}
|
||||
return toActionView(attempt, action, readOutput(action.getOutputJson(), outputType));
|
||||
}
|
||||
|
||||
Instant now = Instant.now();
|
||||
if (action == null) {
|
||||
action = ExperimentAction.running(UUID.randomUUID(), attemptId, stepCode, actionCode, requestFingerprint);
|
||||
repository.insertAction(action);
|
||||
} else if (action.getStatus() == ActionStatus.FAILED || action.getStatus() == ActionStatus.ROLLED_BACK) {
|
||||
action.restart(requestFingerprint, now);
|
||||
repository.saveAction(action);
|
||||
} else {
|
||||
throw validation("该动作正在处理中，请勿重复提交");
|
||||
}
|
||||
|
||||
ActionOutcome<T> outcome = work.run();
|
||||
if (outcome == null) {
|
||||
throw new BusinessException(ErrorCode.INTERNAL_ERROR, "动作未返回执行结果");
|
||||
}
|
||||
|
||||
Instant completedAt = Instant.now();
|
||||
action.complete(outcome.getStatus(), writeOutput(outcome.getOutput()), completedAt);
|
||||
if (outcome.getStatus() == ActionStatus.ROLLED_BACK) {
|
||||
attempt.moveTo(outcome.getNextStepCode(), outcome.getNextActionCode(), completedAt);
|
||||
} else {
|
||||
attempt.moveTo(nextStepCode, nextActionCode, completedAt);
|
||||
}
|
||||
repository.saveAction(action);
|
||||
repository.save(attempt);
|
||||
return toActionView(attempt, action, outcome.getOutput());
|
||||
}
|
||||
|
||||
private ExperimentAttemptView toAttemptView(ExperimentAttempt attempt, List<ExperimentAction> actions) {
|
||||
List<ExperimentActionView<?>> actionViews = new ArrayList<ExperimentActionView<?>>();
|
||||
for (ExperimentAction action : actions) {
|
||||
actionViews.add(toActionView(attempt, action, readOutput(action.getOutputJson(), Object.class)));
|
||||
}
|
||||
return new ExperimentAttemptView(attempt.getId(), attempt.getModule(), attempt.getAttemptNo(),
|
||||
attempt.getStatus(), attempt.getCurrentStepCode(), attempt.getCurrentActionCode(),
|
||||
attempt.getBusinessId(), attempt.getCreatedAt(), attempt.getUpdatedAt(),
|
||||
attempt.getCompletedAt(), actionViews);
|
||||
}
|
||||
|
||||
private <T> ExperimentActionView<T> toActionView(ExperimentAttempt attempt, ExperimentAction action, T output) {
|
||||
return new ExperimentActionView<T>(attempt.getId(), attempt.getModule(), action.getStepCode(),
|
||||
action.getActionCode(), action.getStatus(), attempt.getStatus(), attempt.getCurrentStepCode(),
|
||||
attempt.getCurrentActionCode(), action.getCompletedAt(), output);
|
||||
}
|
||||
|
||||
private String writeOutput(Object output) {
|
||||
try {
|
||||
return output == null ? null : objectMapper.writeValueAsString(output);
|
||||
} catch (JsonProcessingException e) {
|
||||
throw new BusinessException(ErrorCode.INTERNAL_ERROR, "动作结果序列化失败");
|
||||
}
|
||||
}
|
||||
|
||||
private <T> T readOutput(String json, Class<T> outputType) {
|
||||
if (json == null) {
|
||||
return null;
|
||||
}
|
||||
try {
|
||||
return objectMapper.readValue(json, outputType);
|
||||
} catch (JsonProcessingException e) {
|
||||
throw new BusinessException(ErrorCode.INTERNAL_ERROR, "动作结果反序列化失败");
|
||||
}
|
||||
}
|
||||
|
||||
private void requireActive(ExperimentAttempt attempt) {
|
||||
if (attempt.getStatus() != AttemptStatus.IN_PROGRESS && attempt.getStatus() != AttemptStatus.SUBMITTED) {
|
||||
throw validation("实验已结束，不能继续执行动作");
|
||||
}
|
||||
}
|
||||
|
||||
private BusinessException validation(String message) {
|
||||
return new BusinessException(ErrorCode.VALIDATION_ERROR, message);
|
||||
}
|
||||
|
||||
private BusinessException notFound(String message) {
|
||||
return new BusinessException(ErrorCode.RESOURCE_NOT_FOUND, message);
|
||||
}
|
||||
}
|
||||
@ -0,0 +1,48 @@
|
||||
package com.yau.digitalrmb.training.attempt.interfaces.dto;
|
||||
|
||||
import com.yau.digitalrmb.training.attempt.domain.ActionStatus;
|
||||
import com.yau.digitalrmb.training.attempt.domain.AttemptStatus;
|
||||
import com.yau.digitalrmb.training.attempt.domain.ExperimentModule;
|
||||
|
||||
import java.time.Instant;
|
||||
import java.util.UUID;
|
||||
|
||||
public final class ExperimentActionView<T> {
|
||||
private final UUID attemptId;
|
||||
private final ExperimentModule moduleCode;
|
||||
private final String stepCode;
|
||||
private final String actionCode;
|
||||
private final ActionStatus actionStatus;
|
||||
private final AttemptStatus attemptStatus;
|
||||
private final String currentStepCode;
|
||||
private final String currentActionCode;
|
||||
private final Instant completedAt;
|
||||
private final T output;
|
||||
|
||||
public ExperimentActionView(UUID attemptId, ExperimentModule moduleCode, String stepCode, String actionCode,
|
||||
ActionStatus actionStatus, AttemptStatus attemptStatus,
|
||||
String currentStepCode, String currentActionCode,
|
||||
Instant completedAt, T output) {
|
||||
this.attemptId = attemptId;
|
||||
this.moduleCode = moduleCode;
|
||||
this.stepCode = stepCode;
|
||||
this.actionCode = actionCode;
|
||||
this.actionStatus = actionStatus;
|
||||
this.attemptStatus = attemptStatus;
|
||||
this.currentStepCode = currentStepCode;
|
||||
this.currentActionCode = currentActionCode;
|
||||
this.completedAt = completedAt;
|
||||
this.output = output;
|
||||
}
|
||||
|
||||
public UUID getAttemptId() { return attemptId; }
|
||||
public ExperimentModule getModuleCode() { return moduleCode; }
|
||||
public String getStepCode() { return stepCode; }
|
||||
public String getActionCode() { return actionCode; }
|
||||
public ActionStatus getActionStatus() { return actionStatus; }
|
||||
public AttemptStatus getAttemptStatus() { return attemptStatus; }
|
||||
public String getCurrentStepCode() { return currentStepCode; }
|
||||
public String getCurrentActionCode() { return currentActionCode; }
|
||||
public Instant getCompletedAt() { return completedAt; }
|
||||
public T getOutput() { return output; }
|
||||
}
|
||||
@ -0,0 +1,52 @@
|
||||
package com.yau.digitalrmb.training.attempt.interfaces.dto;
|
||||
|
||||
import com.yau.digitalrmb.training.attempt.domain.AttemptStatus;
|
||||
import com.yau.digitalrmb.training.attempt.domain.ExperimentModule;
|
||||
|
||||
import java.time.Instant;
|
||||
import java.util.Collections;
|
||||
import java.util.List;
|
||||
import java.util.UUID;
|
||||
|
||||
public final class ExperimentAttemptView {
|
||||
private final UUID attemptId;
|
||||
private final ExperimentModule moduleCode;
|
||||
private final int attemptNo;
|
||||
private final AttemptStatus attemptStatus;
|
||||
private final String currentStepCode;
|
||||
private final String currentActionCode;
|
||||
private final String businessId;
|
||||
private final Instant createdAt;
|
||||
private final Instant updatedAt;
|
||||
private final Instant completedAt;
|
||||
private final List<ExperimentActionView<?>> actions;
|
||||
|
||||
public ExperimentAttemptView(UUID attemptId, ExperimentModule moduleCode, int attemptNo,
|
||||
AttemptStatus attemptStatus, String currentStepCode, String currentActionCode,
|
||||
String businessId, Instant createdAt, Instant updatedAt, Instant completedAt,
|
||||
List<ExperimentActionView<?>> actions) {
|
||||
this.attemptId = attemptId;
|
||||
this.moduleCode = moduleCode;
|
||||
this.attemptNo = attemptNo;
|
||||
this.attemptStatus = attemptStatus;
|
||||
this.currentStepCode = currentStepCode;
|
||||
this.currentActionCode = currentActionCode;
|
||||
this.businessId = businessId;
|
||||
this.createdAt = createdAt;
|
||||
this.updatedAt = updatedAt;
|
||||
this.completedAt = completedAt;
|
||||
this.actions = actions == null ? Collections.<ExperimentActionView<?>>emptyList() : actions;
|
||||
}
|
||||
|
||||
public UUID getAttemptId() { return attemptId; }
|
||||
public ExperimentModule getModuleCode() { return moduleCode; }
|
||||
public int getAttemptNo() { return attemptNo; }
|
||||
public AttemptStatus getAttemptStatus() { return attemptStatus; }
|
||||
public String getCurrentStepCode() { return currentStepCode; }
|
||||
public String getCurrentActionCode() { return currentActionCode; }
|
||||
public String getBusinessId() { return businessId; }
|
||||
public Instant getCreatedAt() { return createdAt; }
|
||||
public Instant getUpdatedAt() { return updatedAt; }
|
||||
public Instant getCompletedAt() { return completedAt; }
|
||||
public List<ExperimentActionView<?>> getActions() { return actions; }
|
||||
}
|
||||
@ -0,0 +1,103 @@
|
||||
package com.yau.digitalrmb.training.attempt.application;
|
||||
|
||||
import com.yau.digitalrmb.shared.api.ErrorCode;
|
||||
import com.yau.digitalrmb.shared.exception.BusinessException;
|
||||
import com.yau.digitalrmb.training.attempt.domain.ActionStatus;
|
||||
import com.yau.digitalrmb.training.attempt.domain.AttemptStatus;
|
||||
import com.yau.digitalrmb.training.attempt.domain.ExperimentModule;
|
||||
import com.yau.digitalrmb.training.attempt.domain.ExperimentSubject;
|
||||
import com.yau.digitalrmb.training.attempt.interfaces.dto.ExperimentActionView;
|
||||
import com.yau.digitalrmb.training.attempt.interfaces.dto.ExperimentAttemptView;
|
||||
import org.junit.jupiter.api.Test;
|
||||
import org.springframework.beans.factory.annotation.Autowired;
|
||||
import org.springframework.boot.test.context.SpringBootTest;
|
||||
import org.springframework.test.context.ActiveProfiles;
|
||||
import org.springframework.transaction.annotation.Transactional;
|
||||
|
||||
import java.util.concurrent.atomic.AtomicInteger;
|
||||
|
||||
import static org.junit.jupiter.api.Assertions.assertEquals;
|
||||
import static org.junit.jupiter.api.Assertions.assertThrows;
|
||||
|
||||
@SpringBootTest
|
||||
@ActiveProfiles("test")
|
||||
@Transactional
|
||||
class ExperimentAttemptServiceTest {
|
||||
@Autowired
|
||||
private ExperimentAttemptService service;
|
||||
|
||||
private final ExperimentSubject subject = new ExperimentSubject("workflow-user", 101L, 202L, "ORG-W");
|
||||
|
||||
@Test
|
||||
void replaysACompletedActionWithoutRepeatingBusinessWork() {
|
||||
ExperimentAttemptView attempt = service.create(ExperimentModule.EXCHANGE, subject);
|
||||
AtomicInteger calls = new AtomicInteger();
|
||||
|
||||
ExperimentActionView<TestOutput> first = service.execute(attempt.getAttemptId(), subject,
|
||||
"01", "refresh-wallet", "fingerprint-a", TestOutput.class,
|
||||
() -> ActionOutcome.completed(new TestOutput("call-" + calls.incrementAndGet())),
|
||||
"01", "confirm-binding");
|
||||
ExperimentActionView<TestOutput> replay = service.execute(attempt.getAttemptId(), subject,
|
||||
"01", "refresh-wallet", "fingerprint-a", TestOutput.class,
|
||||
() -> ActionOutcome.completed(new TestOutput("call-" + calls.incrementAndGet())),
|
||||
"01", "confirm-binding");
|
||||
|
||||
assertEquals(1, calls.get());
|
||||
assertEquals("call-1", first.getOutput().getValue());
|
||||
assertEquals("call-1", replay.getOutput().getValue());
|
||||
assertEquals(ActionStatus.COMPLETED, replay.getActionStatus());
|
||||
}
|
||||
|
||||
@Test
|
||||
void rejectsDifferentInputForAnAlreadyCompletedAction() {
|
||||
ExperimentAttemptView attempt = service.create(ExperimentModule.PAYMENT, subject);
|
||||
service.execute(attempt.getAttemptId(), subject, "01", "refresh-payer", "fingerprint-a",
|
||||
TestOutput.class, () -> ActionOutcome.completed(new TestOutput("payer")),
|
||||
"01", "refresh-payee");
|
||||
|
||||
BusinessException error = assertThrows(BusinessException.class, () -> service.execute(
|
||||
attempt.getAttemptId(), subject, "01", "refresh-payer", "fingerprint-b",
|
||||
TestOutput.class, () -> ActionOutcome.completed(new TestOutput("changed")),
|
||||
"01", "refresh-payee"));
|
||||
|
||||
assertEquals(ErrorCode.VALIDATION_ERROR, error.getErrorCode());
|
||||
}
|
||||
|
||||
@Test
|
||||
void retriesAnActionAfterCommittedRollback() {
|
||||
ExperimentAttemptView attempt = service.create(ExperimentModule.EXCHANGE, subject);
|
||||
|
||||
ExperimentActionView<TestOutput> rolledBack = service.execute(attempt.getAttemptId(), subject,
|
||||
"05", "reserve-coins", "fingerprint-a", TestOutput.class,
|
||||
() -> ActionOutcome.rolledBack(new TestOutput("released"), "04", "execute-debit-hold"),
|
||||
"05", "adjust-bank-ledger");
|
||||
ExperimentActionView<TestOutput> completed = service.execute(attempt.getAttemptId(), subject,
|
||||
"05", "reserve-coins", "fingerprint-b", TestOutput.class,
|
||||
() -> ActionOutcome.completed(new TestOutput("reserved")),
|
||||
"05", "adjust-bank-ledger");
|
||||
|
||||
assertEquals(ActionStatus.ROLLED_BACK, rolledBack.getActionStatus());
|
||||
assertEquals("04", rolledBack.getCurrentStepCode());
|
||||
assertEquals(ActionStatus.COMPLETED, completed.getActionStatus());
|
||||
assertEquals("reserved", completed.getOutput().getValue());
|
||||
}
|
||||
|
||||
@Test
|
||||
void cancelsAnActiveAttemptAndRemovesItFromCurrentLookup() {
|
||||
ExperimentAttemptView attempt = service.create(ExperimentModule.ISSUANCE, subject);
|
||||
|
||||
ExperimentAttemptView cancelled = service.cancel(attempt.getAttemptId(), subject);
|
||||
|
||||
assertEquals(AttemptStatus.CANCELLED, cancelled.getAttemptStatus());
|
||||
assertThrows(BusinessException.class, () -> service.current(ExperimentModule.ISSUANCE, subject));
|
||||
}
|
||||
|
||||
public static final class TestOutput {
|
||||
private String value;
|
||||
|
||||
public TestOutput() { }
|
||||
public TestOutput(String value) { this.value = value; }
|
||||
public String getValue() { return value; }
|
||||
public void setValue(String value) { this.value = value; }
|
||||
}
|
||||
}
|
||||
Loading…
Reference in New Issue