Dağıtık Transaction — 2PC Neden Nadiren, Saga Neden Sık

1
@Service
2
class CardPaymentSaga {
3

4
    private final AccountClient accounts;
5
    private final CardClient cards;
6
    private final SagaLog log;
7

8
    CardPaymentSaga(AccountClient accounts, CardClient cards, SagaLog log) {
9
        this.accounts = accounts;
10
        this.cards = cards;
11
        this.log = log;
12
    }
13

14
    /** Each step commits on its own; a failure is undone by a new step, not by a rollback. */
15
    PaymentResult pay(PaymentRequest request) {
16
        UUID sagaId = log.start(request);
17
        accounts.debit(request.iban(), request.amount(), "debit-" + sagaId);       // step 1 commits
18
        log.advance(sagaId, Step.DEBITED);
19
        try {
20
            cards.payDebt(request.cardId(), request.amount(), "card-" + sagaId);   // step 2 commits
21
            log.advance(sagaId, Step.COMPLETED);
22
            return PaymentResult.paid(sagaId);
23
        } catch (CardRejectedException rejected) {
24
            accounts.refund(request.iban(), request.amount(), "refund-" + sagaId); // compensation
25
            log.advance(sagaId, Step.COMPENSATED);
26
            return PaymentResult.refunded(sagaId, rejected.reason());
27
        }
28
    }
29
}

1
@Repository
2
class SagaLog {
3

4
    private final JdbcClient jdbc;
5

6
    SagaLog(JdbcClient jdbc) {
7
        this.jdbc = jdbc;
8
    }
9

10
    UUID start(PaymentRequest request) {
11
        UUID id = UUID.randomUUID();
12
        jdbc.sql("INSERT INTO payment_saga (id, iban, card_id, amount, step) VALUES (?, ?, ?, ?, 'STARTED')")
13
            .params(id, request.iban(), request.cardId(), request.amount())
14
            .update();
15
        return id;
16
    }
17

18
    void advance(UUID id, Step step) {
19
        jdbc.sql("UPDATE payment_saga SET step = ?, updated_at = now() WHERE id = ?").params(step.name(), id).update();
20
    }
21

22
    /** Sagas a crash left halfway: a scheduled job resumes or compensates them. */
23
    List<UUID> stuckSince(Instant before) {
24
        return jdbc.sql("SELECT id FROM payment_saga WHERE step = 'DEBITED' AND updated_at < ?")
25
            .param(before).query(UUID.class).list();
26
    }
27
}

1
@Component
2
class AccountClient {
3

4
    private final RestClient rest;
5

6
    AccountClient(RestClient accountsRestClient) {
7
        this.rest = accountsRestClient;
8
    }
9

10
    void debit(String iban, BigDecimal amount, String idempotencyKey) {
11
        post("/accounts/{iban}/debits", iban, amount, idempotencyKey);
12
    }
13

14
    void refund(String iban, BigDecimal amount, String idempotencyKey) {
15
        post("/accounts/{iban}/refunds", iban, amount, idempotencyKey);
16
    }
17

18
    // The same key on every retry: the accounts service applies each debit or refund once.
19
    private void post(String uri, String iban, BigDecimal amount, String key) {
20
        rest.post().uri(uri, iban).header("Idempotency-Key", key).body(new Money(amount)).retrieve().toBodilessEntity();
21
    }
22
}

1
class CardPaymentSagaTest {
2

3
    private final FakeAccounts accounts = new FakeAccounts(Map.of("TR01", new BigDecimal("1000.00")));
4
    private final CardClient rejectingCards = (cardId, amount, key) -> { throw new CardRejectedException("LIMIT"); };
5
    private final CardPaymentSaga saga = new CardPaymentSaga(accounts, rejectingCards, new InMemorySagaLog());
6

7
    @Test
8
    void refundsTheAccountWhenTheCardStepFails() {
9
        PaymentResult result = saga.pay(new PaymentRequest("TR01", "CARD-7", new BigDecimal("250.00")));
10

11
        assertThat(result.refunded()).isTrue();
12
        assertThat(accounts.balance("TR01")).isEqualByComparingTo("1000.00");
13
        assertThat(accounts.history("TR01")).extracting(Entry::type).containsExactly("DEBIT", "REFUND");   // the trace stays
14
    }
15
}