W09 코드 뒤풀이 · 사라진 두 번째 1,000원

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.transfer;

import com.example.financialcore.PostgresIntegrationTestSupport;

import org.junit.jupiter.api.Test;
import org.springframework.aop.support.AopUtils;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.test.context.SpringBootTest;
import org.springframework.transaction.annotation.Transactional;

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

@SpringBootTest
class TransactionProxyIT extends PostgresIntegrationTestSupport {
    @Autowired TransferService transfers;

    @Test
    void transferBeanIsProxiedAndPublicMethodOwnsTheBoundary() throws Exception {
        assertThat(AopUtils.isAopProxy(transfers)).isTrue();
        assertThat(TransferService.class.getMethod("transfer", TransferService.Command.class)
            .isAnnotationPresent(Transactional.class)).isTrue();
    }
}

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.transfer;

import com.example.financialcore.PostgresIntegrationTestSupport;
import com.example.financialcore.account.Account;
import com.example.financialcore.account.AccountOpeningService;
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.boot.test.context.TestConfiguration;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Import;
import org.springframework.jdbc.core.simple.JdbcClient;

import java.util.concurrent.atomic.AtomicBoolean;
import java.util.List;

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

@SpringBootTest
@Import(TransferFailurePointIT.FailureConfiguration.class)
class TransferFailurePointIT extends PostgresIntegrationTestSupport {
    @Autowired JdbcClient jdbc;
    @Autowired AccountOpeningService openings;
    @Autowired TransferService transfers;
    @Autowired AtomicBoolean invoked;

    Account from;
    Account to;

    @BeforeEach void clean() {
        invoked.set(false);
        jdbc.sql("TRUNCATE idempotency_request,ledger_entry,business_tx,account RESTART IDENTITY CASCADE").update();
        from = openings.open("customer-1", "FAIL-FROM", 10_000);
        to = openings.open("customer-2", "FAIL-TO", 5_000);
    }

    @Test
    void runtimeExceptionAfterBusinessMutationRollsBackEveryTransferEffect() {
        assertThatThrownBy(() -> transfers.transfer(new TransferService.Command(
            "customer-1", "W7-FAIL", from.getId(), to.getId(), 1_000)))
            .isInstanceOf(RuntimeException.class).hasMessage("injected after business mutation");
        assertThat(invoked).isTrue();
        long transferTransactions = jdbc.sql("SELECT COUNT(*) FROM business_tx WHERE tx_type='TRANSFER'")
            .query(Long.class).single();
        long transferEntries = jdbc.sql("SELECT COUNT(*) FROM ledger_entry WHERE entry_type LIKE 'TRANSFER_%'")
            .query(Long.class).single();
        assertThat(List.of(
            balance(from.getId()), balance(to.getId()), transferTransactions, transferEntries))
            .as("W7D5_RED_EXPECTED_FULL_ROLLBACK")
            .containsExactly(10_000L, 5_000L, 0L, 0L);
    }

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

    @TestConfiguration(proxyBeanMethods = false)
    static class FailureConfiguration {
        @Bean AtomicBoolean invoked() { return new AtomicBoolean(); }
        @Bean TransferFailureHook failureHook(AtomicBoolean invoked) {
            return () -> {
                invoked.set(true);
                throw new RuntimeException("injected after business mutation");
            };
        }
    }
}

package com.example.financialcore.transfer;

import com.example.financialcore.PostgresIntegrationTestSupport;
import jakarta.persistence.EntityManager;
import jakarta.persistence.PersistenceContext;
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.boot.test.context.TestConfiguration;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Import;
import org.springframework.jdbc.core.simple.JdbcClient;
import org.springframework.transaction.annotation.Propagation;
import org.springframework.transaction.annotation.Transactional;

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

@SpringBootTest
@Import(TransactionPropagationIT.ProbeConfiguration.class)
class TransactionPropagationIT extends PostgresIntegrationTestSupport {
    @TestConfiguration
    static class ProbeConfiguration {
        @Bean InnerProbe innerProbe() { return new InnerProbe(); }
        @Bean OuterProbe outerProbe(InnerProbe inner) { return new OuterProbe(inner); }
    }

    static class InnerProbe {
        @PersistenceContext EntityManager entityManager;

        @Transactional(propagation = Propagation.REQUIRES_NEW)
        public void insertCommittedInnerRow() {
            entityManager.createNativeQuery(
                "INSERT INTO w9_propagation_probe(label) VALUES ('INNER_COMMITTED')")
                .executeUpdate();
        }
    }

    static class OuterProbe {
        @PersistenceContext EntityManager entityManager;
        private final InnerProbe inner;
        OuterProbe(InnerProbe inner) { this.inner = inner; }

        @Transactional
        public void insertOuterThenFail() {
            entityManager.createNativeQuery(
                "INSERT INTO w9_propagation_probe(label) VALUES ('OUTER_ROLLED_BACK')")
                .executeUpdate();
            inner.insertCommittedInnerRow();
            throw new RuntimeException("rollback outer");
        }
    }

    @Autowired JdbcClient jdbc;
    @Autowired OuterProbe outer;

    @BeforeEach
    void prepareProbeTable() {
        jdbc.sql("DROP TABLE IF EXISTS w9_propagation_probe").update();
        jdbc.sql("CREATE TABLE w9_propagation_probe(id BIGINT GENERATED ALWAYS AS IDENTITY, label TEXT NOT NULL)")
            .update();
    }

    @Test
    void requiresNewCommitsWhileTheOuterTransactionRollsBack() {
        assertThatThrownBy(outer::insertOuterThenFail)
            .isInstanceOf(RuntimeException.class)
            .hasMessage("rollback outer");
        assertThat(jdbc.sql("SELECT label FROM w9_propagation_probe ORDER BY id")
            .query(String.class).list()).containsExactly("INNER_COMMITTED");
    }
}

CREATE TABLE account (
    id BIGINT GENERATED BY DEFAULT AS IDENTITY PRIMARY KEY,
    owner_id VARCHAR(64) NOT NULL,
    account_no VARCHAR(32) NOT NULL UNIQUE,
    status VARCHAR(16) NOT NULL CHECK (status IN ('ACTIVE', 'CLOSED')),
    currency VARCHAR(3) NOT NULL CHECK (currency = 'KRW'),
    balance BIGINT NOT NULL CHECK (balance >= 0),
    version BIGINT NOT NULL DEFAULT 0
);

CREATE INDEX idx_account_owner_id ON account(owner_id, id);

CREATE TABLE business_tx (
    id UUID PRIMARY KEY,
    tx_type VARCHAR(32) NOT NULL,
    status VARCHAR(16) NOT NULL,
    correlation_id VARCHAR(64) NOT NULL UNIQUE,
    requested_at TIMESTAMPTZ NOT NULL,
    completed_at TIMESTAMPTZ
);

CREATE TABLE ledger_entry (
    id BIGINT GENERATED BY DEFAULT AS IDENTITY PRIMARY KEY,
    business_tx_id UUID NOT NULL REFERENCES business_tx(id),
    account_id BIGINT NOT NULL REFERENCES account(id),
    entry_type VARCHAR(32) NOT NULL,
    amount BIGINT NOT NULL CHECK (amount > 0),
    signed_amount BIGINT NOT NULL,
    balance_after BIGINT NOT NULL CHECK (balance_after >= 0),
    reversal_of BIGINT REFERENCES ledger_entry(id),
    created_at TIMESTAMPTZ NOT NULL,
    UNIQUE (business_tx_id, account_id, entry_type)
);

CREATE INDEX idx_ledger_account_created_id
    ON ledger_entry(account_id, created_at DESC, id DESC);

CREATE TABLE idempotency_request (
    id BIGINT GENERATED BY DEFAULT AS IDENTITY PRIMARY KEY,
    scope VARCHAR(64) NOT NULL,
    actor_id VARCHAR(64) NOT NULL,
    idempotency_key VARCHAR(128) NOT NULL,
    request_hash CHAR(64) NOT NULL,
    status VARCHAR(16) NOT NULL,
    response_status INTEGER,
    response_body TEXT,
    created_at TIMESTAMPTZ NOT NULL,
    completed_at TIMESTAMPTZ,
    UNIQUE (scope, actor_id, idempotency_key)
);

package com.example.financialcore;

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 static org.assertj.core.api.Assertions.assertThat;

@SpringBootTest
class CoreSchemaIT extends PostgresIntegrationTestSupport {
    @Autowired JdbcClient jdbc;

    @Test
    void v001CreatesExactlyTheRequiredCoreTables() {
        var tables = jdbc.sql("""
                SELECT table_name FROM information_schema.tables
                WHERE table_schema='public' AND table_type='BASE TABLE'
                  AND table_name <> 'flyway_schema_history'
                ORDER BY table_name
                """).query(String.class).list();
        assertThat(tables)
            .as("W6D1_RED_EXPECTED_FOUR_CORE_TABLES")
            .containsExactlyInAnyOrder("account", "business_tx", "ledger_entry", "idempotency_request");
    }
}