W10 코드 뒤풀이 · 기다릴까, 충돌로 멈출까

package com.example.financialcore.transfer;

import com.example.financialcore.PostgresIntegrationTestSupport;
import com.example.financialcore.account.Account;
import com.example.financialcore.account.AccountOpeningService;
import com.example.financialcore.account.AccountRepository;
import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.test.context.SpringBootTest;
import org.springframework.jdbc.core.simple.JdbcClient;

import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.Executors;
import java.util.concurrent.Future;
import java.util.concurrent.TimeUnit;

import static org.assertj.core.api.Assertions.assertThat;

@SpringBootTest
class SortedLockTransferIT extends PostgresIntegrationTestSupport {
    @Autowired JdbcClient jdbc;
    @Autowired AccountOpeningService openings;
    @Autowired AccountRepository accounts;
    @Autowired TransferService transfers;
    Account first;
    Account second;

    @BeforeEach
    void setUp() {
        jdbc.sql("TRUNCATE idempotency_request,ledger_entry,business_tx,account RESTART IDENTITY CASCADE")
            .update();
        first = openings.open("customer-1", "SORT-A", 10_000);
        second = openings.open("customer-1", "SORT-B", 10_000);
    }

    @Test
    void oppositeDirectionsFinishWithoutDeadlockAndPreserveTotal() throws Exception {
        int pairs = 10;
        int tasks = pairs * 2;
        var ready = new CountDownLatch(tasks);
        var start = new CountDownLatch(1);
        var pool = Executors.newFixedThreadPool(tasks);
        try {
            List<Future<?>> futures = new ArrayList<>();
            for (int i = 0; i < pairs; i++) {
                int sequence = i;
                futures.add(pool.submit(() -> invoke(ready, start, "ab-" + sequence,
                    first.getId(), second.getId())));
                futures.add(pool.submit(() -> invoke(ready, start, "ba-" + sequence,
                    second.getId(), first.getId())));
            }
            assertThat(ready.await(10, TimeUnit.SECONDS)).isTrue();
            start.countDown();
            for (Future<?> future : futures) future.get(30, TimeUnit.SECONDS);
            long firstBalance = accounts.findById(first.getId()).orElseThrow().getBalance();
            long secondBalance = accounts.findById(second.getId()).orElseThrow().getBalance();
            assertThat(firstBalance + secondBalance).isEqualTo(20_000);
            assertThat(jdbc.sql("SELECT COALESCE(SUM(signed_amount),0) FROM ledger_entry WHERE entry_type LIKE 'TRANSFER_%'")
                .query(Long.class).single()).isZero();
        } finally {
            start.countDown();
            pool.shutdownNow();
        }
    }

    private Object invoke(
        CountDownLatch ready, CountDownLatch start, String key, long from, long to
    ) {
        try {
            ready.countDown();
            start.await();
            return transfers.transfer(new TransferService.Command("customer-1", key, from, to, 100));
        } catch (InterruptedException interrupted) {
            Thread.currentThread().interrupt();
            throw new IllegalStateException(interrupted);
        }
    }
}

package com.example.financialcore.account;

import com.example.financialcore.PostgresIntegrationTestSupport;
import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.test.context.SpringBootTest;
import org.springframework.dao.OptimisticLockingFailureException;
import org.springframework.jdbc.core.simple.JdbcClient;
import org.springframework.transaction.PlatformTransactionManager;
import org.springframework.transaction.TransactionDefinition;
import org.springframework.transaction.support.TransactionTemplate;

import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.Executors;
import java.util.concurrent.Future;
import java.util.concurrent.TimeUnit;

import static org.assertj.core.api.Assertions.assertThat;

@SpringBootTest
class OptimisticAccountIT extends PostgresIntegrationTestSupport {
    enum Outcome { COMMIT, OPTIMISTIC_FAILURE, TECHNICAL_FAILURE }

    @Autowired JdbcClient jdbc;
    @Autowired AccountOpeningService openings;
    @Autowired AccountRepository accounts;
    @Autowired PlatformTransactionManager transactionManager;

    private long accountId;

    @BeforeEach
    void setUp() {
        jdbc.sql("TRUNCATE idempotency_request, ledger_entry, business_tx, account RESTART IDENTITY CASCADE")
            .update();
        accountId = openings.open("OPTIMISTIC", 10_000).getId();
    }

    @Test
    void production_versioned_entity_has_one_commit_and_one_optimistic_failure() throws Exception {
        var readBarrier = new CountDownLatch(2);
        var pool = Executors.newFixedThreadPool(2);
        List<Future<Outcome>> futures = new ArrayList<>();
        try {
            for (int i = 0; i < 2; i++) {
                futures.add(pool.submit(() -> runWithdrawal(readBarrier)));
            }
            List<Outcome> outcomes = new ArrayList<>();
            for (Future<Outcome> future : futures) outcomes.add(future.get(20, TimeUnit.SECONDS));
            assertThat(outcomes).containsExactlyInAnyOrder(Outcome.COMMIT, Outcome.OPTIMISTIC_FAILURE);
            assertThat(jdbc.sql("SELECT balance FROM account WHERE id=:id")
                .param("id", accountId).query(Long.class).single()).isEqualTo(9_000);
            assertThat(jdbc.sql("SELECT version FROM account WHERE id=:id")
                .param("id", accountId).query(Long.class).single()).isEqualTo(1);
        } finally {
            pool.shutdownNow();
        }
    }

    private Outcome runWithdrawal(CountDownLatch readBarrier) {
        try {
            TransactionTemplate tx = new TransactionTemplate(transactionManager);
            tx.setPropagationBehavior(TransactionDefinition.PROPAGATION_REQUIRES_NEW);
            tx.setTimeout(10);
            tx.executeWithoutResult(status -> {
                Account account = accounts.findById(accountId).orElseThrow();
                readBarrier.countDown();
                await(readBarrier);
                account.withdraw(1_000);
                accounts.flush();
            });
            return Outcome.COMMIT;
        } catch (OptimisticLockingFailureException expected) {
            return Outcome.OPTIMISTIC_FAILURE;
        } catch (RuntimeException failure) {
            if (hasOptimisticCause(failure)) return Outcome.OPTIMISTIC_FAILURE;
            return Outcome.TECHNICAL_FAILURE;
        }
    }

    private static boolean hasOptimisticCause(Throwable failure) {
        for (Throwable cause = failure; cause != null; cause = cause.getCause()) {
            if (cause.getClass().getSimpleName().contains("OptimisticLock")) return true;
        }
        return false;
    }

    private static void await(CountDownLatch latch) {
        try {
            if (!latch.await(10, TimeUnit.SECONDS)) throw new IllegalStateException("read barrier timeout");
        } catch (InterruptedException interrupted) {
            Thread.currentThread().interrupt();
            throw new IllegalStateException("interrupted", interrupted);
        }
    }
}

package com.example.financialcore.account;

import com.example.financialcore.PostgresIntegrationTestSupport;
import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.test.context.SpringBootTest;
import org.springframework.jdbc.core.simple.JdbcClient;
import org.springframework.transaction.PlatformTransactionManager;
import org.springframework.transaction.TransactionDefinition;
import org.springframework.transaction.support.TransactionTemplate;

import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.Executors;
import java.util.concurrent.Future;
import java.util.concurrent.TimeUnit;

import static org.assertj.core.api.Assertions.assertThat;

@SpringBootTest
class LostUpdateBaselineIT extends PostgresIntegrationTestSupport {
    @Autowired JdbcClient jdbc;
    @Autowired AccountOpeningService openings;
    @Autowired PlatformTransactionManager transactionManager;

    private long accountId;

    @BeforeEach
    void setUp() {
        jdbc.sql("TRUNCATE idempotency_request, ledger_entry, business_tx, account RESTART IDENTITY CASCADE")
            .update();
        accountId = openings.open("LOST-UPDATE", 10_000).getId();
    }

    @Test
    void test_only_versionless_unconditional_updates_commit_twice_and_lose_one_update() throws Exception {
        var readBarrier = new CountDownLatch(2);
        var pool = Executors.newFixedThreadPool(2);
        List<Future<Boolean>> futures = new ArrayList<>();
        try {
            for (int i = 0; i < 2; i++) {
                futures.add(pool.submit(() -> {
                    TransactionTemplate tx = requiresNew();
                    tx.executeWithoutResult(status -> {
                        long observed = jdbc.sql("SELECT balance FROM account WHERE id=:id")
                            .param("id", accountId).query(Long.class).single();
                        readBarrier.countDown();
                        await(readBarrier);
                        jdbc.sql("UPDATE account SET balance=:balance WHERE id=:id")
                            .param("balance", observed - 1_000)
                            .param("id", accountId)
                            .update();
                    });
                    return true;
                }));
            }
            int commits = 0;
            for (Future<Boolean> future : futures) if (future.get(20, TimeUnit.SECONDS)) commits++;
            assertThat(commits).isEqualTo(2);
            assertThat(balance()).isEqualTo(9_000);
            assertThat(version()).isZero();
        } finally {
            pool.shutdownNow();
        }
    }

    @Test
    void production_pessimistic_lock_path_commits_twice_and_preserves_both_updates() throws Exception {
        var start = new CountDownLatch(1);
        var pool = Executors.newFixedThreadPool(2);
        List<Future<Boolean>> futures = new ArrayList<>();
        try {
            for (int i = 0; i < 2; i++) {
                futures.add(pool.submit(() -> {
                    start.await();
                    TransactionTemplate tx = requiresNew();
                    tx.executeWithoutResult(status -> {
                        long observed = jdbc.sql("SELECT balance FROM account WHERE id=:id FOR UPDATE")
                            .param("id", accountId).query(Long.class).single();
                        jdbc.sql("UPDATE account SET balance=:balance, version=version+1 WHERE id=:id")
                            .param("balance", observed - 1_000)
                            .param("id", accountId)
                            .update();
                    });
                    return true;
                }));
            }
            start.countDown();
            int commits = 0;
            for (Future<Boolean> future : futures) if (future.get(20, TimeUnit.SECONDS)) commits++;
            assertThat(commits).isEqualTo(2);
            assertThat(balance()).isEqualTo(8_000);
        } finally {
            start.countDown();
            pool.shutdownNow();
        }
    }

    private TransactionTemplate requiresNew() {
        TransactionTemplate tx = new TransactionTemplate(transactionManager);
        tx.setPropagationBehavior(TransactionDefinition.PROPAGATION_REQUIRES_NEW);
        tx.setTimeout(10);
        return tx;
    }

    private long balance() {
        return jdbc.sql("SELECT balance FROM account WHERE id=:id")
            .param("id", accountId).query(Long.class).single();
    }

    private long version() {
        return jdbc.sql("SELECT version FROM account WHERE id=:id")
            .param("id", accountId).query(Long.class).single();
    }

    private static void await(CountDownLatch latch) {
        try {
            if (!latch.await(10, TimeUnit.SECONDS)) throw new IllegalStateException("read barrier timeout");
        } catch (InterruptedException interrupted) {
            Thread.currentThread().interrupt();
            throw new IllegalStateException("interrupted", interrupted);
        }
    }
}

package com.example.financialcore.account;

import jakarta.persistence.LockModeType;
import org.springframework.data.jpa.repository.JpaRepository;
import org.springframework.data.jpa.repository.Lock;
import org.springframework.data.jpa.repository.Query;
import org.springframework.data.repository.query.Param;

import java.util.List;
import java.util.Optional;

public interface AccountRepository extends JpaRepository<Account, Long> {
    Optional<Account> findByAccountNo(String accountNo);

    boolean existsByAccountNo(String accountNo);

    @Query("select a.ownerId from Account a where a.id = :id")
    Optional<String> findOwnerId(@Param("id") long id);

    @Lock(LockModeType.PESSIMISTIC_WRITE)
    @Query("select a from Account a where a.id = :id")
    Optional<Account> findOneForUpdate(@Param("id") long id);

    @Lock(LockModeType.PESSIMISTIC_WRITE)
    @Query("select a from Account a where a.id in :ids order by a.id")
    List<Account> findAllForUpdateOrderById(@Param("ids") List<Long> ids);
}

package com.example.financialcore.transfer;

import com.example.financialcore.account.Account;
import com.example.financialcore.account.AccountRepository;
import com.example.financialcore.api.BusinessException;
import com.example.financialcore.api.ErrorCode;
import com.example.financialcore.ledger.BusinessTransaction;
import com.example.financialcore.ledger.BusinessTransactionRepository;
import com.example.financialcore.ledger.LedgerEntry;
import com.example.financialcore.ledger.LedgerEntryRepository;
import org.springframework.beans.factory.ObjectProvider;
import org.springframework.stereotype.Service;
import org.springframework.transaction.annotation.Transactional;

import java.time.Instant;
import java.util.HashMap;
import java.util.List;

@Service
public class TransferService {
    public record Command(String actorId, String transactionId, long fromAccountId, long toAccountId, long amount) {}
    public record Result(String businessTransactionId, long fromBalance, long toBalance) {}

    private final AccountRepository accounts;
    private final BusinessTransactionRepository transactions;
    private final LedgerEntryRepository ledger;
    private final TransferFailureHook failureHook;

    public TransferService(
        AccountRepository accounts, BusinessTransactionRepository transactions,
        LedgerEntryRepository ledger, ObjectProvider<TransferFailureHook> hooks
    ) {
        this.accounts = accounts;
        this.transactions = transactions;
        this.ledger = ledger;
        this.failureHook = hooks.getIfAvailable(() -> TransferFailureHook.NONE);
    }

    @Transactional
    public Result transfer(Command command) {
        validate(command);
        var locked = accounts.findAllForUpdateOrderById(
            List.of(command.fromAccountId(), command.toAccountId()).stream().sorted().toList());
        if (locked.size() != 2) throw new BusinessException(ErrorCode.ACCOUNT_NOT_FOUND, "account not found");
        var byId = new HashMap<Long, Account>();
        locked.forEach(account -> byId.put(account.getId(), account));
        Account from = byId.get(command.fromAccountId());
        Account to = byId.get(command.toAccountId());
        if (!from.getOwnerId().equals(command.actorId())) throw new BusinessException(ErrorCode.ACCESS_DENIED, "access denied");
        from.withdraw(command.amount());
        to.deposit(command.amount());
        Instant now = Instant.now();
        var tx = transactions.save(BusinessTransaction.completedTransfer(command.transactionId(), now));
        ledger.save(LedgerEntry.transferOut(tx, from, command.amount(), now));
        ledger.save(LedgerEntry.transferIn(tx, to, command.amount(), now));
        failureHook.afterBusinessMutation();
        return new Result(tx.getId().toString(), from.getBalance(), to.getBalance());
    }

    private static void validate(Command command) {
        if (command.actorId() == null || command.actorId().isBlank()) throw new IllegalArgumentException("actorId");
        if (command.transactionId() == null || command.transactionId().isBlank()) throw new IllegalArgumentException("transactionId");
        if (command.fromAccountId() <= 0 || command.toAccountId() <= 0) throw new IllegalArgumentException("account id");
        if (command.fromAccountId() == command.toAccountId()) throw new IllegalArgumentException("same account");
        if (command.amount() <= 0) throw new IllegalArgumentException("amount");
    }
}

package com.example.financialcore.account;

import com.example.financialcore.api.BusinessException;
import com.example.financialcore.api.ErrorCode;

import jakarta.persistence.Column;
import jakarta.persistence.Entity;
import jakarta.persistence.EnumType;
import jakarta.persistence.Enumerated;
import jakarta.persistence.GeneratedValue;
import jakarta.persistence.GenerationType;
import jakarta.persistence.Id;
import jakarta.persistence.Table;
import jakarta.persistence.Version;

@Entity
@Table(name = "account")
public class Account {
    public enum Status { ACTIVE, CLOSED }

    @Id
    @GeneratedValue(strategy = GenerationType.IDENTITY)
    private Long id;

    @Column(name = "account_no", nullable = false, unique = true, length = 32)
    private String accountNo;

    @Column(name = "owner_id", nullable = false, length = 64)
    private String ownerId;

    @Enumerated(EnumType.STRING)
    @Column(nullable = false, length = 16)
    private Status status;

    @Column(nullable = false, length = 3)
    private String currency;

    @Column(nullable = false)
    private long balance;

    @Version
    private long version;

    protected Account() {}

    private Account(String ownerId, String accountNo, long initialBalance) {
        if (ownerId == null || ownerId.isBlank()) throw new IllegalArgumentException("ownerId");
        if (accountNo == null || accountNo.isBlank()) throw new IllegalArgumentException("accountNo");
        if (initialBalance < 0) throw new IllegalArgumentException("initialBalance");
        this.ownerId = ownerId;
        this.accountNo = accountNo;
        this.status = Status.ACTIVE;
        this.currency = "KRW";
        this.balance = initialBalance;
    }

    public static Account open(String accountNo, long initialBalance) {
        return new Account("system", accountNo, initialBalance);
    }

    public static Account open(String ownerId, String accountNo, long initialBalance) {
        return new Account(ownerId, accountNo, initialBalance);
    }

    public void deposit(long amount) {
        requireActive();
        if (amount <= 0) throw new IllegalArgumentException("amount");
        this.balance = Math.addExact(this.balance, amount);
    }

    public void withdraw(long amount) {
        requireActive();
        if (amount <= 0) throw new IllegalArgumentException("amount");
        if (balance < amount) {
            throw new BusinessException(ErrorCode.INSUFFICIENT_BALANCE, "insufficient balance");
        }
        this.balance -= amount;
    }

    private void requireActive() {
        if (status != Status.ACTIVE) throw new IllegalStateException("account is not active");
    }

    public Long getId() { return id; }
    public String getAccountNo() { return accountNo; }
    public String getOwnerId() { return ownerId; }
    public Status getStatus() { return status; }
    public String getCurrency() { return currency; }
    public long getBalance() { return balance; }
    public long getVersion() { return version; }
}