Saga Pattern

1
@Entity
2
public class CardPaymentSaga {
3

4
    // Pay a credit card bill from a current account: two services, two databases.
5
    public enum Step { HOLDING_FUNDS, PAYING_CARD, CONFIRMED, COMPENSATING, CANCELLED }
6

7
    @Id
8
    private UUID id;
9
    private long paymentId;
10

11
    @Enumerated(EnumType.STRING)
12
    private Step step;
13

14
    private Instant stepStartedAt;
15
    private String failureReason;
16

17
    @Version
18
    private long version;          // two replies racing for the same saga: one wins, one retries
19

20
    public static CardPaymentSaga start(long paymentId, Clock clock) {
21
        var saga = new CardPaymentSaga();
22
        saga.id = UUID.randomUUID();
23
        saga.paymentId = paymentId;
24
        saga.moveTo(Step.HOLDING_FUNDS, clock);
25
        return saga;
26
    }
27

28
    public void moveTo(Step next, Clock clock) {
29
        this.step = next;
30
        this.stepStartedAt = Instant.now(clock);
31
    }
32

33
    public void fail(String reason, Clock clock) {
34
        this.failureReason = reason;
35
        moveTo(Step.COMPENSATING, clock);
36
    }
37

38
    // getters omitted
39
}

1
@Service
2
class CardPaymentOrchestrator {
3

4
    private final SagaRepository sagas;
5
    private final SagaCommands commands;
6
    private final Clock clock;
7

8
    CardPaymentOrchestrator(SagaRepository sagas, SagaCommands commands, Clock clock) {
9
        this.sagas = sagas;
10
        this.commands = commands;
11
        this.clock = clock;
12
    }
13

14
    @Transactional
15
    public UUID begin(long paymentId) {
16
        CardPaymentSaga saga = sagas.save(CardPaymentSaga.start(paymentId, clock));
17
        commands.holdFunds(saga.getId(), paymentId);        // via outbox, same transaction
18
        return saga.getId();
19
    }
20

21
    @Transactional
22
    public void fundsHeld(UUID sagaId) {
23
        CardPaymentSaga saga = sagas.findById(sagaId).orElseThrow();
24
        if (saga.getStep() != CardPaymentSaga.Step.HOLDING_FUNDS) return;   // duplicate or late reply
25
        saga.moveTo(CardPaymentSaga.Step.PAYING_CARD, clock);
26
        commands.payCard(sagaId, saga.getPaymentId());
27
    }
28

29
    @Transactional
30
    public void cardPaid(UUID sagaId) {
31
        CardPaymentSaga saga = sagas.findById(sagaId).orElseThrow();
32
        if (saga.getStep() != CardPaymentSaga.Step.PAYING_CARD) return;
33
        saga.moveTo(CardPaymentSaga.Step.CONFIRMED, clock);
34
        commands.confirmPayment(saga.getPaymentId());       // the hold becomes a real debit
35
    }
36

37
    @Transactional
38
    public void stepFailed(UUID sagaId, String reason) {
39
        CardPaymentSaga saga = sagas.findById(sagaId).orElseThrow();
40
        CardPaymentSaga.Step failedAt = saga.getStep();
41
        if (failedAt == CardPaymentSaga.Step.CONFIRMED || failedAt == CardPaymentSaga.Step.CANCELLED) return;
42
        saga.fail(reason, clock);
43
        // Undo only what already happened, newest first.
44
        if (failedAt == CardPaymentSaga.Step.PAYING_CARD) commands.releaseHold(sagaId, saga.getPaymentId());
45
        commands.cancelPayment(saga.getPaymentId(), reason);
46
        saga.moveTo(CardPaymentSaga.Step.CANCELLED, clock);
47
    }
48
}

1
@Component
2
class SagaReplies {
3

4
    private final CardPaymentOrchestrator orchestrator;
5

6
    SagaReplies(CardPaymentOrchestrator orchestrator) {
7
        this.orchestrator = orchestrator;
8
    }
9

10
    @KafkaListener(topics = "saga-replies", groupId = "card-payment-saga")
11
    public void on(SagaReply reply) {
12
        switch (reply) {
13
            case SagaReply.FundsHeld r -> orchestrator.fundsHeld(r.sagaId());
14
            case SagaReply.CardPaid r -> orchestrator.cardPaid(r.sagaId());
15
            case SagaReply.StepFailed r -> orchestrator.stepFailed(r.sagaId(), r.reason());
16
        }
17
    }
18
}
19

20
sealed interface SagaReply {
21
    UUID sagaId();
22
    record FundsHeld(UUID sagaId) implements SagaReply {}
23
    record CardPaid(UUID sagaId) implements SagaReply {}
24
    record StepFailed(UUID sagaId, String reason) implements SagaReply {}
25
}

1
@Component
2
class SagaCommands {
3

4
    private final OutboxRepository outbox;
5

6
    SagaCommands(OutboxRepository outbox) {
7
        this.outbox = outbox;
8
    }
9

10
    // Commands go through the outbox: saga state and outgoing command commit together.
11
    void holdFunds(UUID sagaId, long paymentId)   { send("accounts", "HoldFunds", sagaId, paymentId); }
12
    void payCard(UUID sagaId, long paymentId)     { send("cards", "PayCardBalance", sagaId, paymentId); }
13
    void confirmPayment(long paymentId)           { send("accounts", "CaptureHold", null, paymentId); }
14

15
    // Compensations are new forward actions, not rollbacks.
16
    void releaseHold(UUID sagaId, long paymentId) { send("accounts", "ReleaseHold", sagaId, paymentId); }
17
    void cancelPayment(long paymentId, String reason) { send("payments", "CancelPayment", null, paymentId); }
18

19
    private void send(String target, String type, UUID sagaId, long paymentId) {
20
        outbox.save(OutboxEvent.command(target, type, sagaId, paymentId));
21
    }
22
}

1
@Component
2
class SagaTimeouts {
3

4
    private final SagaRepository sagas;
5
    private final CardPaymentOrchestrator orchestrator;
6
    private final Clock clock;
7

8
    SagaTimeouts(SagaRepository sagas, CardPaymentOrchestrator orchestrator, Clock clock) {
9
        this.sagas = sagas;
10
        this.orchestrator = orchestrator;
11
        this.clock = clock;
12
    }
13

14
    // A reply that never comes must not leave the customer's money on hold forever.
15
    @Scheduled(fixedDelay = 30_000)
16
    @SchedulerLock(name = "sagaTimeouts", lockAtMostFor = "PT5M")
17
    public void failStuckSagas() {
18
        Instant cutoff = Instant.now(clock).minus(Duration.ofMinutes(10));
19
        for (UUID id : sagas.findStuckSince(cutoff, List.of(CardPaymentSaga.Step.HOLDING_FUNDS, CardPaymentSaga.Step.PAYING_CARD))) {
20
            orchestrator.stepFailed(id, "timeout");
21
        }
22
    }
23
}